use std::net::SocketAddr;
use std::sync::Arc;
use anyhow::{bail, Context, Result};
use gunnar_core::observe::{MetricsSnapshot, RequestId, ServerMetrics};
use gunnar_core::{ensure_host_key, Readiness};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_ssh::{BoundServer, GunnarSsh, SshConfig};
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use gunnar_common::bindspec::BindSpec;
use crate::control_plane::CoreControlPlane;
use crate::stores::ServerStores;
use crate::transport::{WireConfig, WireTransport};
#[derive(Debug, Clone)]
pub struct ServeConfig {
pub ssh_bind: SocketAddr,
pub grpc_bind: BindSpec,
pub grpc_bind_device: Option<String>,
pub admin_key: Option<String>,
pub restore_pubkeys: Vec<String>,
pub no_grpc: bool,
pub wg_only: bool,
pub twin_role: bool,
pub allow_anonymous: bool,
pub anonymous: gunnar_ssh::AnonymousPolicy,
#[cfg(feature = "http")]
pub http_bind: Option<SocketAddr>,
#[cfg(feature = "http")]
pub http_cleartext: gunnar_http::Cleartext,
#[cfg(feature = "http")]
pub http_tls: Option<HttpTls>,
pub runner: Option<crate::runner::WatchConfig>,
pub grpc_tls: Option<GrpcTls>,
pub web_relying_parties: Vec<String>,
pub web_require_user_verification: bool,
pub identity_root_site: bool,
pub grpc_replication_bind: Option<SocketAddr>,
pub max_concurrent_transfers: Option<usize>,
pub memory_budget_bytes: Option<u64>,
pub admission_wait: Option<std::time::Duration>,
pub max_transfers_per_principal: Option<usize>,
pub max_anonymous_transfers: Option<usize>,
#[cfg(feature = "admit-share")]
pub admit: gunnar_ssh::edda_admit::Settings,
pub per_principal_limit_is_hard: bool,
pub control_plane_workers: Option<usize>,
pub max_object_bytes: Option<u64>,
pub metrics_interval: Option<std::time::Duration>,
pub metrics_bind: Option<SocketAddr>,
#[cfg(feature = "replication")]
pub replication: Option<crate::mirror::ReplicationConfig>,
#[cfg(feature = "replication")]
pub standby_watch: Option<crate::standby_watch::StandbyWatch>,
#[cfg(feature = "replication")]
pub repair: Option<crate::repair::RepairConfig>,
pub replication_view: crate::replication_view::ReplicationView,
pub tenants_view: crate::dashboard::TenantView,
pub flow_view: crate::dashboard::FlowView,
pub wire: WireConfig,
#[cfg(feature = "metered")]
pub metered: Option<crate::entitlement::MeteredConfig>,
pub disk_floor: gunnar_ssh::DiskFloor,
pub space_probe: Option<Arc<dyn gunnar_ssh::SpaceProbe>>,
}
pub use gunnar_ssh::AnonymousPolicy;
impl Default for ServeConfig {
fn default() -> Self {
ServeConfig {
ssh_bind: SocketAddr::from(([0, 0, 0, 0], gunnar_ssh::config::DEFAULT_PORT)),
grpc_bind: BindSpec::Addr(gunnar_grpc::DEFAULT_BIND),
grpc_bind_device: None,
admin_key: None,
restore_pubkeys: Vec::new(),
no_grpc: false,
wg_only: false,
twin_role: false,
allow_anonymous: true,
anonymous: gunnar_ssh::AnonymousPolicy::Allow,
#[cfg(feature = "http")]
http_bind: None,
#[cfg(feature = "http")]
http_cleartext: gunnar_http::Cleartext::default(),
#[cfg(feature = "http")]
http_tls: None,
grpc_tls: None,
web_relying_parties: Vec::new(),
identity_root_site: false,
web_require_user_verification: false,
grpc_replication_bind: None,
runner: None,
max_concurrent_transfers: None,
memory_budget_bytes: None,
admission_wait: None,
max_transfers_per_principal: None,
max_anonymous_transfers: None,
#[cfg(feature = "admit-share")]
admit: gunnar_ssh::edda_admit::Settings::default(),
per_principal_limit_is_hard: false,
control_plane_workers: None,
max_object_bytes: None,
metrics_interval: Some(METRICS_INTERVAL),
metrics_bind: None,
#[cfg(feature = "replication")]
replication: None,
#[cfg(feature = "replication")]
standby_watch: None,
#[cfg(feature = "replication")]
repair: None,
replication_view: crate::replication_view::ReplicationView::new(),
tenants_view: crate::dashboard::TenantView::new(),
flow_view: crate::dashboard::FlowView::new(),
wire: WireConfig::default(),
#[cfg(feature = "metered")]
metered: None,
disk_floor: gunnar_ssh::DiskFloor::default(),
space_probe: None,
}
}
}
impl ServeConfig {
pub fn ephemeral() -> Self {
ServeConfig {
ssh_bind: SocketAddr::from(([127, 0, 0, 1], 0)),
grpc_bind: BindSpec::Addr(SocketAddr::from(([127, 0, 0, 1], 0))),
grpc_bind_device: None,
..ServeConfig::default()
}
}
}
pub const GRPC_SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(2);
pub const METRICS_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60);
pub const GIT_BLOCKING_THREADS: usize = 512;
pub fn control_plane_workers() -> usize {
let cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(4);
(cores / 4).clamp(4, 8)
}
pub const CONTROL_CONNECTION_WORKERS: usize = 2;
fn cpus_allowed() -> Option<usize> {
let status = std::fs::read_to_string("/proc/self/status").ok()?;
let list = status
.lines()
.find_map(|line| line.strip_prefix("Cpus_allowed_list:"))?;
count_cpu_list(list)
}
fn count_cpu_list(list: &str) -> Option<usize> {
let mut total = 0usize;
for range in list.trim().split(',') {
let range = range.trim();
if range.is_empty() {
continue;
}
total += match range.split_once('-') {
Some((lo, hi)) => {
let (lo, hi) = (
lo.trim().parse::<usize>().ok()?,
hi.trim().parse::<usize>().ok()?,
);
hi.checked_sub(lo)?.checked_add(1)?
}
None => {
range.parse::<usize>().ok()?;
1
}
};
}
(total > 0).then_some(total)
}
fn spread_budget(allowed: Option<usize>, quota: Option<f64>) -> Option<(usize, usize)> {
let (allowed, quota) = (allowed?, quota?);
let budget = quota.ceil().max(1.0) as usize;
(allowed >= budget.saturating_mul(2)).then_some((allowed, budget))
}
fn cpu_quota() -> Option<f64> {
let own = std::fs::read_to_string("/proc/self/cgroup").ok()?;
let rel = own
.lines()
.find_map(|line| line.strip_prefix("0::"))?
.trim()
.trim_start_matches('/');
let path = std::path::Path::new("/sys/fs/cgroup")
.join(rel)
.join("cpu.max");
let raw = std::fs::read_to_string(path).ok()?;
let mut parts = raw.split_whitespace();
let quota = parts.next()?;
if quota == "max" {
return None;
}
let quota: f64 = quota.parse().ok()?;
let period: f64 = parts.next()?.parse().ok()?;
(period > 0.0).then(|| quota / period)
}
fn report_cpu_budget() {
let Some((allowed, budget)) = spread_budget(cpus_allowed(), cpu_quota()) else {
return;
};
tracing::warn!(
cpus_allowed = allowed,
cpu_quota = budget,
"this server may run on {allowed} CPUs but may only use {budget} CPUs' worth of \
them. Every hand-off between the three threads that serve one transfer then \
crosses a core, and in a 2026-08-04 measurement that cost 2.7x the CPU per \
request at 32 concurrent clients against the identical budget applied as a \
cpuset. Start the \
container with `--cpuset-cpus` (or the unit with `AllowedCPUs=`) naming {budget} \
CPUs, not only `--cpus`/`CPUQuota=`."
);
}
pub fn runtime() -> Result<tokio::runtime::Runtime> {
tokio::runtime::Builder::new_multi_thread()
.max_blocking_threads(GIT_BLOCKING_THREADS)
.thread_name("gunnar-git")
.enable_all()
.build()
.context("building the git runtime")
}
#[cfg(feature = "http")]
#[derive(Debug, Clone)]
pub enum HttpTls {
SelfMinted,
Identity(gunnar_http::TlsIdentity),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GrpcTls {
SelfMinted,
}
#[derive(Default)]
struct ControlRuntime(Vec<tokio::runtime::Runtime>);
impl ControlRuntime {
fn new(runtime: tokio::runtime::Runtime) -> Self {
ControlRuntime(vec![runtime])
}
fn hold(&mut self, other: &mut ControlRuntime) {
self.0.append(&mut other.0);
}
fn handle(&self) -> Option<tokio::runtime::Handle> {
self.0.first().map(|runtime| runtime.handle().clone())
}
fn shutdown(&mut self) {
for runtime in self.0.drain(..) {
runtime.shutdown_background();
}
}
}
impl Drop for ControlRuntime {
fn drop(&mut self) {
self.shutdown();
}
}
pub struct Server {
ssh: Option<BoundServer>,
grpc: Option<JoinHandle<Result<()>>>,
grpc_shutdown: Option<oneshot::Sender<()>>,
grpc_endpoint: Option<(SocketAddr, bool)>,
control_runtime: ControlRuntime,
stores: Arc<ServerStores>,
bootstrap_window_open: bool,
host_key_fingerprint: String,
metrics: Arc<ServerMetrics>,
heartbeat: Option<JoinHandle<()>>,
#[cfg(feature = "replication")]
standby_watch: Option<JoinHandle<()>>,
#[cfg(feature = "replication")]
repair: Option<JoinHandle<()>>,
bus: Arc<EventBus>,
#[cfg(feature = "http")]
http: Option<(JoinHandle<std::io::Result<()>>, SocketAddr, bool)>,
scrape: Option<(JoinHandle<std::io::Result<()>>, SocketAddr)>,
#[cfg(feature = "replication")]
replication: Arc<crate::mirror::ReplicationSlot>,
#[cfg(feature = "webhooks")]
webhook: Arc<crate::webhook::WebhookSlot>,
}
async fn heartbeat(metrics: Arc<ServerMetrics>, interval: std::time::Duration) {
let mut ticker = tokio::time::interval(interval);
ticker.tick().await;
let mut previous = metrics.snapshot();
loop {
ticker.tick().await;
let now = metrics.snapshot();
if now == previous {
continue;
}
emit_metrics(&now);
previous = now;
}
}
fn grpc_request_id() -> u64 {
RequestId::next().get()
}
fn emit_metrics(now: &MetricsSnapshot) {
now.emit_heartbeat();
}
impl Server {
pub async fn start(stores: Arc<ServerStores>, config: ServeConfig) -> Result<Server> {
report_cpu_budget();
let mut config = config;
config.grpc_bind = crate::boot::role_bound_here(config.grpc_bind)?;
if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
config.grpc_bind_device = Some(interface.clone());
}
if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
let has_address =
|i: &str| matches!(appliance_pid1::net::interface_ipv4(i), Ok(Some(_)));
if !has_address(interface) && appliance_pid1::net::dhcp_pending(interface) {
tracing::info!(interface = %interface, "the control plane waits for {interface}'s address (PID 1 asks DHCP again in the background)");
while !has_address(interface) && appliance_pid1::net::dhcp_pending(interface) {
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
}
}
}
#[cfg(feature = "wg-client")]
if !config.wg_only {
if let Ok(Some(seed)) = crate::pairing::seed(stores.root()) {
if crate::tunnel_addr::estate_is_wg_only(Some(
seed.tunnel == nordisk_pairing::seed::TunnelKind::Wireguard,
)) {
config.wg_only = true;
tracing::warn!("serve: this box belongs to an estate (a WireGuard pairing seed) — WireGuard-only: its utility address serves nothing, boxes speak over the tunnel");
}
}
}
crate::tunnel_addr::set_wg_only(config.wg_only);
let resolved_grpc_bind = match resolve_grpc_bind(&config.grpc_bind)? {
Some(addr) => addr,
None => {
config.no_grpc = true;
gunnar_grpc::DEFAULT_BIND
}
};
let admin_key = match &config.admin_key {
Some(spec) => Some(read_key_argument(spec)?),
None => None,
};
let restore_keys = config
.restore_pubkeys
.iter()
.map(|spec| {
let line = read_key_argument(spec)?;
gunnar_core::auth::parse_public_key(line.trim()).map_err(|err| {
anyhow::anyhow!(
"--restore-pubkey {spec:?} is not an OpenSSH public key line: {err:#}"
)
})?;
Ok(line.trim().to_owned())
})
.collect::<Result<Vec<String>>>()?;
let readiness = stores
.auth()
.prepare_to_serve(admin_key.as_deref())
.with_context(|| {
format!(
"reading the identity table at {}. This is NOT the same as finding it \
empty: a data path that cannot be read must never be served as a server \
nobody has enrolled yet, because that would reopen the bootstrap window on \
a disk fault.",
stores.root().display()
)
})?;
announce_readiness(
&readiness,
stores.root(),
resolved_grpc_bind,
config.no_grpc,
);
let bootstrap_window_open = readiness.bootstrap_window_open();
let host_key_store = if stores.is_twin() {
stores.secrets_unframed()?
} else {
stores.secrets()?
};
let host_key =
ensure_host_key(&host_key_store).context("loading or minting the SSH host key")?;
if stores.is_twin() {
crate::twin::start_follower(Arc::clone(&stores));
}
#[cfg(feature = "wg-client")]
match crate::pairing::seed(stores.root()) {
Ok(Some(seed)) => crate::pairing::start(Arc::clone(&stores), seed),
Ok(None) => {}
Err(e) => {
tracing::error!(error = %format!("{e:#}"), "pairing: the seed is present and unusable — this box will not pair")
}
}
let host_key_fingerprint = host_key.fingerprint().to_string();
tracing::info!(
fingerprint = %host_key_fingerprint,
"ssh host key ready (persisted; it survives a restart)"
);
let russh_key = russh_private_key(host_key.private_openssh())
.context("handing the host key to russh")?;
match stores.pki() {
Ok(pki) => match pki.issuer() {
Ok(Some(live)) => tracing::info!(
subject = %live.subject,
fingerprint = %live.fingerprint,
"issuer ready (persisted; it survives a restart)"
),
Ok(None) => match pki.create("gunnar issuer", gunnar_core::DEFAULT_ISSUER_DAYS) {
Ok(created) => tracing::warn!(
subject = %created.subject,
fingerprint = %created.fingerprint,
"FIRST START: this server minted its own certificate issuer. Its \
private key is sealed in the secret store. `gunnar issuer show --pem` \
prints the certificate clients pin."
),
Err(err) => tracing::warn!(
%err,
"the issuer could not be minted at first start; certificate issuance \
is unavailable until `gunnar issuer create` or `issuer import` succeeds"
),
},
Err(err) => tracing::warn!(%err, "the PKI document could not be read"),
},
Err(err) => tracing::warn!(
%err,
"no secret store, so no issuer: certificate issuance is unavailable"
),
}
if !config.no_grpc && config.grpc_tls.is_some() {
let preflight = crate::tls_preflight::preflight(&stores)
.context("checking whether this server can move to mutual TLS")?;
if let Some(refusal) = preflight.refusal() {
anyhow::bail!("{refusal}");
}
preflight.warn();
}
let metrics = Arc::clone(stores.metrics());
let bus = Arc::new(EventBus::new(gunnar_grpc::DEFAULT_EVENT_BUFFER));
let transport = Arc::new(
WireTransport::new(Arc::clone(&stores))
.with_bus(Arc::clone(&bus))
.with_metrics(Arc::clone(&metrics))
.with_wire_config(config.wire.clone()),
);
let defaults = SshConfig::default();
let mut budgets = crate::budget::budgets();
let twin_shaped = stores.is_twin() || config.twin_role;
crate::twin::note_git_bind_is_twin_shaped(twin_shaped);
let ssh_config = SshConfig {
bind: if twin_shaped {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, asked = %config.ssh_bind, "this box is a TWIN: the git listener is bound to loopback only");
SocketAddr::from(([127, 0, 0, 1], 0))
} else {
config.ssh_bind
},
allow_anonymous: config.allow_anonymous,
server_id: gunnar_common::ssh_server_id(),
max_concurrent_transfers: config.max_concurrent_transfers.or(Some(budgets.transfers)),
memory_budget_bytes: config.memory_budget_bytes.or(Some(budgets.memory_bytes)),
admission_wait: config.admission_wait.or(defaults.admission_wait),
max_transfers_per_principal: config
.max_transfers_per_principal
.or(defaults.max_transfers_per_principal),
max_anonymous_transfers: config
.max_anonymous_transfers
.or(defaults.max_anonymous_transfers),
per_principal_limit_is_hard: config.per_principal_limit_is_hard,
#[cfg(feature = "admit-share")]
admit: config.admit,
..defaults
};
if let Some(n) = config.max_concurrent_transfers {
budgets.transfers = n;
budgets.transfers_from = crate::budget::Source::Operator;
}
if let Some(bytes) = config.memory_budget_bytes {
budgets.memory_bytes = bytes;
}
if let Some(wait) = config.admission_wait {
budgets.wait_ms = wait.as_millis() as u64;
}
crate::budget::announce(&budgets);
#[cfg(feature = "http")]
let http_transport: Arc<dyn gunnar_ssh::seam::GitTransport> = Arc::clone(&transport) as _;
#[cfg(feature = "metered")]
let entitlement = match &config.metered {
Some(metered) => Some(
crate::entitlement::EntitlementSlot::open(Arc::clone(&stores), metered)
.context("opening the entitlement slot")?,
),
None => {
tracing::warn!(
"metered build started without --monetize-pubkey: every namespace is Free \
and Entitlement.Set refuses every fact"
);
None
}
};
#[cfg(feature = "metered")]
let sells_room = entitlement
.as_ref()
.map(|slot| slot.sells_room_flag())
.unwrap_or_else(|| Arc::new(std::sync::atomic::AtomicBool::new(false)));
#[cfg(not(feature = "metered"))]
let sells_room = Arc::new(std::sync::atomic::AtomicBool::new(false));
nordisk_pairing::feedbus::on_published("the front's egress door", crate::feed::take_egress);
{
let flag = Arc::clone(&sells_room);
nordisk_pairing::feedbus::on_published(
"shop room: a person's account with no fact",
move |feed| {
let sells_room = crate::feed::Fed::from_feed(feed).sells_room;
if flag.swap(sells_room, std::sync::atomic::Ordering::Relaxed) != sells_room {
tracing::info!(target: "gunnar::audit", event = "shop_room", sells_room, "{}", if sells_room { "this estate's room comes from the shop: a person's own account with no fact holds no room until its fact lands" } else { "this estate's room does not come from a shop: an account with no fact is uncapped" });
}
},
);
}
#[cfg(feature = "metered")]
if let Some(slot) = &entitlement {
let taker = Arc::clone(slot);
crate::feed::on_monetize_key(
"entitlement slot",
Box::new(move |k| taker.set_key(k).map(|_| ())),
);
let pay = Arc::clone(slot);
nordisk_pairing::feedbus::on_published(
"entitlement: the site's pay page",
move |feed| {
if let Some(url) = crate::feed::Fed::from_feed(feed).billing_url.as_deref() {
if let Err(why) = pay.set_billing_url(url) {
tracing::warn!(target: "gunnar::audit", event = "entitlement.billing_url_refused", reason = %why, "the fed pay page was refused; the one in force stays");
}
}
},
);
let slot = Arc::clone(slot);
config.tenants_view.fill(move || slot.counts());
}
let (mut swept, mut swept_bytes) =
crate::push_space::sweep_spool_leftovers(stores.repos_root());
let (lfs, lfs_bytes) = crate::push_space::sweep_lfs_spool(
&stores.txlog().dir().join(gunnar_core::txlog::SPOOL_DIR),
);
swept += lfs;
swept_bytes += lfs_bytes;
if swept > 0 {
tracing::warn!(
files = swept,
bytes = swept_bytes,
"deleted {swept} pushed-pack spool file(s) a crash left behind ({})",
gunnar_ssh::headroom::human(swept_bytes)
);
}
let (fs_slack, unmeasured) = gunnar_ssh::space::FsSlack::of(stores.root());
match &unmeasured {
None => {
tracing::info!(root = %stores.root().display(), measured_permille = ?fs_slack.measured_permille, slack_permille = fs_slack.slack_permille, "{}", fs_slack.words(stores.root(), None))
}
Some(why) => {
tracing::warn!(root = %stores.root().display(), slack_permille = fs_slack.slack_permille, "{}", fs_slack.words(stores.root(), Some(why)))
}
}
let ledger = match &config.space_probe {
None => gunnar_ssh::SpaceLedger::new(stores.root(), config.disk_floor),
Some(probe) => gunnar_ssh::SpaceLedger::with_probe(
stores.root(),
config.disk_floor,
Arc::clone(probe),
),
};
let disk_gate = Arc::new(gunnar_ssh::DiskHeadroomGate::over(Arc::new(
ledger.with_fs_slack(fs_slack),
)));
match disk_gate.headroom() {
Ok(room) => tracing::info!(
root = %stores.root().display(),
free = %gunnar_ssh::headroom::human(room.free),
floor = %gunnar_ssh::headroom::human(room.floor),
disk_floor = %config.disk_floor,
has_room = room.has_room(),
"disk headroom: pushes and Repos.Create are refused under the floor"
),
Err(err) => tracing::error!(
root = %stores.root().display(),
%err,
"disk headroom could not be measured; every push and Repos.Create will be \
refused until it can be"
),
}
let front_control = || {
front_control(
&stores,
config.anonymous.clone(),
Arc::clone(&disk_gate),
#[cfg(feature = "metered")]
entitlement.as_ref(),
)
};
#[cfg(feature = "http")]
let http_lfs: Arc<dyn gunnar_ssh::seam::LfsTransport> = Arc::clone(&transport) as _;
let repo_door_ctx: Arc<std::sync::OnceLock<gunnar_grpc::service::ServiceContext>> =
Arc::new(std::sync::OnceLock::new());
let ssh = GunnarSsh::new(ssh_config, front_control(), transport)
.with_readiness(crate::readiness::flag())
.with_flow(Arc::clone(stores.flow()))
.with_metrics(Arc::clone(&metrics));
#[cfg(feature = "wg-client")]
let ssh = {
let root = stores.root().to_path_buf();
ssh.with_tunnel_address_door(Arc::new(move |address: &str| {
crate::appliance::tunnel_address_answer(&root, address, true)
.map(|answer| crate::appliance::render_tunnel_answer(&answer))
.map_err(|err| format!("{err}"))
}))
};
let ssh = {
let cell = Arc::clone(&repo_door_ctx);
ssh.with_repo_door(Arc::new(move |ask: gunnar_ssh::RepoAsk| {
let cell = Arc::clone(&cell);
Box::pin(async move {
let Some(ctx) = cell.get() else {
return Err(
"this server answers repository settings only on its control plane"
.to_owned(),
);
};
let setting = match ask.action {
gunnar_ssh::RepoAction::Show => gunnar_grpc::repo_door::RepoSetting::Show,
gunnar_ssh::RepoAction::Visibility(public) => {
gunnar_grpc::repo_door::RepoSetting::Visibility(public)
}
gunnar_ssh::RepoAction::DefaultBranch(b) => {
gunnar_grpc::repo_door::RepoSetting::DefaultBranch(b)
}
gunnar_ssh::RepoAction::Create { sha256 } => {
gunnar_grpc::repo_door::RepoSetting::Create(sha256)
}
};
gunnar_grpc::repo_door::ask(
ctx,
&ask.principal,
&ask.fingerprint,
ask.repo.namespace(),
ask.repo.name(),
setting,
)
.await
.map(|repo| gunnar_grpc::repo_door::render(&repo))
.map_err(|status| status.message().to_owned())
})
}))
};
let ssh = if config.wg_only && !config.no_grpc {
let utility = resolved_grpc_bind.ip();
tracing::warn!(%utility, "ssh: --wg-only — a connection at the utility address is refused; boxes speak over the tunnel");
ssh.with_refused_local(Arc::new(move |local: std::net::IpAddr| local == utility))
} else {
ssh
};
let ssh = ssh
.bind(russh_key)
.await
.with_context(|| format!("binding the SSH listener on {}", config.ssh_bind))?;
tracing::info!(
addr = %ssh.local_addr(),
anonymous_read = config.allow_anonymous,
"gunnar ssh listener: git-upload-pack, git-receive-pack; publickey, plus \
anonymous read of public repositories when enabled"
);
#[cfg(feature = "http")]
let http = match config.http_bind {
None => None,
Some(bind) => {
let tls = match &config.http_tls {
None => None,
Some(HttpTls::Identity(identity)) => Some(identity.clone()),
Some(HttpTls::SelfMinted) => {
let pki = stores.pki().context(
"--http-tls-self needs the secret store; no PKI is available",
)?;
let issuer = pki
.issuer()
.context("reading the issuer for --http-tls-self")?
.context(
"--http-tls-self needs an issuer and none exists; the first-start \
mint above must have failed — its warning says why",
)?;
let mut hosts = vec!["localhost".to_owned(), "127.0.0.1".to_owned()];
let ip = bind.ip();
if !ip.is_unspecified() && !ip.is_loopback() {
hosts.push(ip.to_string());
}
let issued = pki
.issue_server_identity(&hosts, gunnar_core::DEFAULT_LEAF_DAYS)
.context("minting the smart-HTTP server certificate")?;
let key_pem = issued.key_pem.clone().context(
"the issuer minted a server certificate but returned no key",
)?;
tracing::info!(
fingerprint = %issued.fingerprint,
sans = ?hosts,
"smart-HTTP TLS: server certificate minted from this server's own \
issuer; clients pin the issuer (`gunnar issuer show --pem`)"
);
Some(gunnar_http::TlsIdentity {
cert_chain_pem: format!("{}{}", issued.cert_pem, issuer.cert_pem),
key_pem,
})
}
};
let http_config = gunnar_http::HttpConfig {
listen: bind,
allow_anonymous: config.allow_anonymous,
cleartext: config.http_cleartext,
tls,
..gunnar_http::HttpConfig::default()
};
let bound = gunnar_http::GunnarHttp::new(
http_config,
gunnar_ssh::server::Dispatch::new(front_control()),
Arc::clone(&http_transport),
)
.with_metrics(Arc::clone(&metrics))
.with_admission(ssh.admission_gate())
.with_lfs(Arc::clone(&http_lfs))
.with_registry(Arc::new(crate::registry::StoreRegistry::new(Arc::clone(
&stores,
))))
.bind()
.await
.with_context(|| format!("binding the HTTP listener on {bind}"))?;
let addr = bound.local_addr();
let tls_on = bound.tls();
tracing::info!(
%addr,
anonymous_read = config.allow_anonymous,
cleartext_credentials = ?config.http_cleartext,
tls = tls_on,
"gunnar smart-http listener: info/refs, git-upload-pack, git-receive-pack, \
info/lfs/objects (batch, get, put, verify)"
);
Some((tokio::spawn(bound.serve()), addr, tls_on))
}
};
let scrape = match config.metrics_bind {
None => None,
Some(bind) => {
let bound = crate::scrape::bind(bind, Arc::clone(&metrics))
.await
.with_context(|| format!("binding the metrics listener on {bind}"))?;
let addr = bound.local_addr();
Some((tokio::spawn(bound.serve()), addr))
}
};
#[cfg(feature = "replication")]
let replication = crate::mirror::ReplicationSlot::new(
Arc::clone(&stores),
Arc::clone(&bus),
tokio::runtime::Handle::current(),
);
#[cfg(feature = "webhooks")]
let webhook = crate::webhook::WebhookSlot::new(
Arc::clone(&stores),
Arc::clone(&bus),
tokio::runtime::Handle::current(),
);
let mut control_view: Option<crate::replication_view::ReplicationView> = None;
#[cfg(feature = "replication")]
let mut control_for_repair: Option<Arc<dyn gunnar_grpc::ControlPlane>> = None;
#[cfg(feature = "replication")]
let mut grow_for_repair: Option<Arc<crate::grow::GrowMachine>> = None;
let (grpc, grpc_shutdown, grpc_endpoint, control_runtime) = if config.no_grpc {
tracing::warn!(
"the gRPC control plane is disabled; repositories, keys and grants can only be \
administered from this host's command line"
);
(None, None, None, ControlRuntime::default())
} else {
let workers = config
.control_plane_workers
.unwrap_or_else(control_plane_workers);
let mut handler_runtime = ControlRuntime::new(
tokio::runtime::Builder::new_multi_thread()
.worker_threads(workers)
.thread_name("gunnar-control-call")
.enable_all()
.build()
.context("building the control plane's handler runtime")?,
);
let handlers = handler_runtime
.handle()
.context("the control plane's handler runtime was gone as soon as it was built")?;
let control_runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(CONTROL_CONNECTION_WORKERS)
.thread_name("gunnar-control")
.enable_all()
.build()
.context("building the control plane's runtime")?;
let grpc_bind = resolved_grpc_bind;
let grpc_bind_device = config.grpc_bind_device.clone();
let web_relying_parties = config
.web_relying_parties
.iter()
.map(|party| {
party
.parse::<gunnar_grpc::webauthn::RelyingParty>()
.map_err(|why| anyhow::anyhow!("--web-relying-party: {why}"))
})
.collect::<Result<Vec<_>>>()?;
let ssh_hint = ssh.local_addr().to_string();
let heartbeats = Arc::new(gunnar_grpc::heartbeat::Registry::default());
#[cfg(feature = "grow-runtime")]
let grow_world: Arc<dyn crate::grow::World> = Arc::new(
crate::grow::world::BoxWorld::new(ssh.admission_gate(), Arc::clone(&stores))
.with_replication(Arc::clone(&replication))
.with_heartbeats(Arc::clone(&heartbeats)),
);
#[cfg(not(feature = "grow-runtime"))]
let grow_world: Arc<dyn crate::grow::World> = Arc::new(crate::grow::world::NoEngine);
#[cfg(feature = "metered")]
let grow = match &config.metered {
Some(metered) => {
let policy = Arc::new(match &metered.monetize_pubkey {
Some(key) => crate::grow::MonetizePolicy::new(key)
.map_err(|e| anyhow::anyhow!("the growth go-ahead policy: {e}"))?,
None => crate::grow::MonetizePolicy::awaiting(),
});
let taker = Arc::clone(&policy);
crate::feed::on_monetize_key(
"growth go-ahead policy",
Box::new(move |k| taker.set_key(k)),
);
crate::grow::GrowMachine::with_policy(grow_world, policy)
}
None => crate::grow::GrowMachine::new(grow_world),
};
#[cfg(not(feature = "metered"))]
let grow = crate::grow::GrowMachine::new(grow_world);
#[cfg(feature = "replication")]
{
grow_for_repair = Some(Arc::clone(&grow));
}
let mut plane = CoreControlPlane::new(Arc::clone(&stores))
.with_restore_keys(restore_keys.clone())
.with_headroom(Arc::clone(&disk_gate))
.with_grow(grow)
.with_admission(ssh.admission_gate())
.with_bus(Arc::clone(&bus))
.with_twin_role(config.twin_role)
.with_git_listener(crate::readiness::GitListener {
asked: config.ssh_bind,
bound: ssh.local_addr(),
});
#[cfg(feature = "replication")]
{
plane = plane.with_replication(Arc::clone(&replication));
}
#[cfg(feature = "webhooks")]
{
plane = plane.with_webhooks(Arc::clone(&webhook));
}
#[cfg(feature = "metered")]
if let Some(slot) = &entitlement {
plane = plane.with_entitlement(Arc::clone(slot));
}
let plane_view = plane.replication_view();
if let Some(bytes) = config.max_object_bytes {
plane = plane.with_max_object_bytes(bytes);
}
if !config.twin_role {
let _on = control_runtime.enter();
let resumed = plane.resume_interrupted_mirrors();
if resumed > 0 {
tracing::info!(target: "gunnar::git", resumed, "mirror syncs interrupted by the last stop were resumed");
}
}
let grpc_trust = match config.grpc_tls {
None => None,
Some(GrpcTls::SelfMinted) => {
let pki = stores
.pki()
.context("--grpc-tls-self needs the secret store; no PKI is available")?;
let issuer = pki
.issuer()
.context("reading the issuer for --grpc-tls-self")?
.context(
"--grpc-tls-self needs an issuer and none exists; the first-start \
mint above must have failed — its warning says why",
)?;
let mut hosts = vec!["localhost".to_owned(), "127.0.0.1".to_owned()];
let ip = grpc_bind.ip();
if !ip.is_unspecified() && !ip.is_loopback() {
hosts.push(ip.to_string());
}
let issued = pki
.issue_server_identity(&hosts, gunnar_core::DEFAULT_LEAF_DAYS)
.context("minting the gRPC server certificate")?;
let key_pem = issued
.key_pem
.clone()
.context("the issuer minted a server certificate but returned no key")?;
tracing::info!(
fingerprint = %issued.fingerprint,
sans = ?hosts,
"gRPC control plane: mutual TLS on. Clients must present a certificate \
this server's issuer signed AND registered (`gunnar control bootstrap`, \
`gunnar key add`)"
);
Some(gunnar_grpc::Trust {
#[cfg(feature = "wg-client")]
client_ca: crate::fed_principals::grpc_client_cas(
&issuer.cert_pem,
stores.root(),
)
.into_bytes(),
#[cfg(not(feature = "wg-client"))]
client_ca: issuer.cert_pem.clone().into_bytes(),
cert: format!("{}{}", issued.cert_pem, issuer.cert_pem).into_bytes(),
key: key_pem.into_bytes(),
})
}
};
let grpc_mutual_tls = grpc_trust.is_some();
let replication_root = stores.root().to_path_buf();
let replication = {
let derived = (!grpc_bind.ip().is_loopback() && grpc_bind.port() != 0)
.then(|| {
gunnar_grpc::client_signer::replication_address(&grpc_bind.to_string()).ok()
})
.flatten()
.and_then(|at| at.parse::<SocketAddr>().ok());
let explicit = config.grpc_replication_bind.is_some();
match config.grpc_replication_bind.or(derived) {
None => None,
Some(base) => {
let bind = base;
let pki = stores.pki().with_context(|| {
format!("the mutual replication listener on {bind} needs the secret store; no PKI is available")
})?;
let issuer = pki.issuer().context("reading the issuer for the replication listener")?.with_context(|| {
format!(
"the mutual replication listener on {bind} needs this box's issuer and there is \
none — the first-start mint above must have failed, and its warning says why. \
Without it no twin of this box can ever follow it."
)
})?;
let mut hosts = vec![
gunnar_grpc::client_signer::REPLICATION_SERVER_NAME.to_owned(),
"localhost".to_owned(),
"127.0.0.1".to_owned(),
];
for ip in [bind.ip(), base.ip()] {
if !ip.is_unspecified()
&& !ip.is_loopback()
&& !hosts.contains(&ip.to_string())
{
hosts.push(ip.to_string());
}
}
let issued = pki
.issue_server_identity(&hosts, gunnar_core::DEFAULT_LEAF_DAYS)
.context("minting the replication listener's certificate")?;
let key_pem = issued.key_pem.clone().context(
"the issuer minted the replication certificate but returned no key",
)?;
Some((
base,
explicit,
bind,
gunnar_grpc::Trust {
client_ca: issuer.cert_pem.clone().into_bytes(),
cert: format!("{}{}", issued.cert_pem, issuer.cert_pem)
.into_bytes(),
key: key_pem.into_bytes(),
},
))
}
}
};
let control = Arc::new(plane);
control_view = Some(plane_view);
#[cfg(feature = "replication")]
{
control_for_repair = Some(control.clone());
}
let bus_for_grpc = Arc::clone(&bus);
let grpc_transfer_gate = crate::grpc_gate::control_transfer_gate(ssh.admission_gate());
let browse_flag = stores.browse();
let audit_ledger = {
let path = stores.audit_path();
let originates = !stores.is_twin() && !crate::twin::is_restoring(stores.root());
let opened = if originates {
match gunnar_core::AuditChain::open_framed(&path) {
Ok(chain) => Ok(chain),
Err(err @ gunnar_core::AuditError::Unframed { .. }) => {
tracing::error!(
target: gunnar_grpc::AUDIT_TARGET,
path = %path.display(),
error = %err,
"this box's audit trail is NOT replicated: it will be written \
here and it will NOT survive a restore from the twin. Every \
other class of state comes back; this one will not."
);
gunnar_core::AuditChain::open(&path)
}
Err(err) => Err(err),
}
} else {
gunnar_core::AuditChain::open(&path)
};
match opened {
Ok(chain) => Some(crate::grpc_gate::control_audit_ledger(Arc::new(chain))),
Err(err) => {
tracing::error!(
target: gunnar_grpc::AUDIT_TARGET,
path = %path.display(),
error = %err,
"the control-plane audit chain could not be opened; this server \
will log its decisions and will NOT write them to a trail. If the \
chain is reported BROKEN, the file has been edited or truncated \
since gunnar wrote it."
);
None
}
}
};
let (tx, rx) = oneshot::channel();
let (bound_tx, bound_rx) = oneshot::channel::<Result<(SocketAddr, bool)>>();
if let Err(e) = gunnar_common::jobs::set_ledger(&stores.root().join("jobs")) {
tracing::error!("ALERT jobs.ledger {e}: the jobs run, unrecorded");
}
let sweep_stores = Arc::clone(&stores);
let activity_stores = Arc::clone(&stores);
let wg_only = config.wg_only;
let join = control_runtime.spawn(async move {
let listener = match bind_when_the_address_arrives(
|| bind_grpc_listener(grpc_bind, grpc_bind_device.as_deref()),
GRPC_BIND_PATIENCE,
GRPC_BIND_RETRY_EVERY,
)
.await
.0
.with_context(|| match &grpc_bind_device {
Some(dev) => format!("binding the gRPC listener on {grpc_bind} (device {dev})"),
None => format!("binding the gRPC listener on {grpc_bind}"),
}) {
Ok(listener) => listener,
Err(err) => {
let _ = bound_tx.send(Err(err));
return Ok(());
}
};
let addr = match listener
.local_addr()
.context("reading the gRPC local address")
{
Ok(addr) => addr,
Err(err) => {
let _ = bound_tx.send(Err(err));
return Ok(());
}
};
let server_config = ServerConfig::default()
.with_bind(addr)
.with_ssh_bind_hint(ssh_hint)
.with_web_relying_parties(web_relying_parties)
.with_web_user_verification(if config.web_require_user_verification {
gunnar_grpc::config::UserVerification::Required
} else {
gunnar_grpc::config::UserVerification::Preferred
})
.with_mutual_tls(grpc_mutual_tls);
gunnar_grpc::warn_authentication_disabled(&server_config);
let authenticated = server_config.authentication_enabled();
let ctx = ServiceContext::new(control, bus_for_grpc, server_config)
.with_sells_room(Arc::clone(&sells_room))
.with_handler_runtime(handlers)
.with_transfer_gate(grpc_transfer_gate)
.with_browse_flag(browse_flag)
.with_heartbeats(Arc::clone(&heartbeats));
let ctx = ctx.with_audit_ledger(crate::grpc_gate::activity_ledger(
audit_ledger,
Arc::clone(activity_stores.activity()),
));
let identity = match crate::identity::try_build(
activity_stores.root(),
Arc::clone(&ctx.control),
&ctx.config,
config.identity_root_site,
) {
Ok(identity) => identity,
Err(why) => {
let _ = bound_tx.send(Err(anyhow::anyhow!("{why}")));
return Ok(());
}
};
activity_stores.set_identity(&identity);
let ctx = ctx.with_identity(identity);
let _ = repo_door_ctx.set(ctx.clone());
let sso_loopback = std::env::var("GUNNAR_SSO_ALLOW_LOOPBACK_HTTP").is_ok_and(|v| v.trim() == "1");
if sso_loopback {
tracing::warn!("GUNNAR_SSO_ALLOW_LOOPBACK_HTTP=1: an identity provider on this box's loopback is reachable over http (a test door)");
}
ctx.sso.set_allow_loopback_http(sso_loopback);
#[cfg(feature = "sso-egress")]
ctx.sso.set_egress(std::sync::Arc::new(crate::sso_egress::UreqEgress::with_nets(
sso_loopback,
crate::sso_egress::system_lookup(),
ctx.sso.private_nets_handle(),
)));
#[cfg(not(feature = "sso-egress"))]
tracing::info!("company SSO has no way out in this build (gunnar-server's `sso-egress` is off): an identity provider's call is refused by name");
{
let env = std::env::var("GUNNAR_SSO_EGRESS").ok();
let on = crate::feed::sso_egress_on(crate::feed::Fed::current().as_ref(), env.as_deref());
ctx.sso.set_egress_on(on);
tracing::info!(target: "gunnar::audit", event = "sso.egress", on, "company SSO's network is {}", if on { "ON: account admins' identity providers may be dialled" } else { "off (airgap): no identity provider is dialled" });
let sso = std::sync::Arc::clone(&ctx.sso);
let resolver_env = std::env::var("GUNNAR_SSO_DNS_RESOLVER").ok();
if let Some(resolver) = crate::feed::sso_dns_resolver(crate::feed::Fed::current().as_ref(), resolver_env.as_deref()) {
ctx.sso.set_dns_resolver(&resolver);
}
nordisk_pairing::feedbus::on_published("sso: the operator's egress switch", move |feed| {
let fed = crate::feed::Fed::from_feed(feed);
sso.set_dns_resolver(crate::feed::sso_dns_resolver(Some(&fed), resolver_env.as_deref()).as_deref().unwrap_or(""));
let on = crate::feed::sso_egress_on(Some(&fed), env.as_deref());
if on != sso.egress_on() {
tracing::info!(target: "gunnar::audit", event = "sso.egress", on, "company SSO's network turned {}", if on { "ON" } else { "off" });
sso.set_egress_on(on);
}
});
}
{
let sso = std::sync::Arc::clone(&ctx.sso);
nordisk_pairing::feedbus::on_published(
"sso: the operator's identity-provider networks",
move |feed| {
let nets = match crate::feed::sso_private_nets(Some(&crate::feed::Fed::from_feed(feed))) {
Ok(nets) => nets,
Err(why) => {
tracing::warn!(target: "gunnar::audit", event = "sso.private_nets_refused", reason = %why, "the fed identity-provider networks were refused; none are admitted");
gunnar_grpc::sso::PrivateNets::default()
}
};
if nets != sso.private_nets() {
tracing::info!(target: "gunnar::audit", event = "sso.private_nets", nets = %nets, "identity providers on the operator's internal networks: {}", if nets.is_empty() { "none (public addresses only)".to_owned() } else { nets.to_string() });
sso.set_private_nets(nets);
}
},
);
}
tokio::spawn(gunnar_grpc::retention::run(ctx.clone()));
if let Err(e) = crate::account_audit_sweep::start(&sweep_stores, crate::account_audit_sweep::EVERY) {
tracing::error!(target: "gunnar::audit", "ALERT {} no sweep ({e}): the account audit logs are not pruned on this run", crate::account_audit_sweep::SWEEP_JOB);
}
let (replication_stop_tx, replication_stop_rx) = tokio::sync::watch::channel(false);
match replication {
Some((base, explicit, bind, trust)) => {
let listener = match tokio::net::TcpListener::bind(bind).await {
Ok(listener) => listener,
Err(err) => {
let _ = bound_tx.send(Err(anyhow::Error::new(err).context(format!(
"binding the mutual replication listener on {bind} (the door this box's twin follows it through)"
))));
return Ok(());
}
};
let bound = listener.local_addr().unwrap_or(base);
let recorded = if wg_only { record_replication_door_wg_only(&replication_root, bound.port(), None) } else { record_replication_door(&replication_root, bound, None) };
if let Err(err) = recorded {
let _ = bound_tx.send(Err(err));
return Ok(());
}
if let Err(err) = gunnar_grpc::router_with_trust(ctx.clone(), grpc_request_id, Some(&trust)) {
let _ = bound_tx.send(Err(err.context("configuring the mutual replication listener")));
return Ok(());
}
tracing::info!(
%bind, suite = gunnar_grpc::preferred_bulk_suite(),
"mutual replication listener up: TLS 1.3, a client certificate from this box's issuer required"
);
tokio::spawn(replication_door(
listener,
base,
explicit,
wg_only,
ctx.clone(),
grpc_request_id,
trust,
replication_root.clone(),
replication_stop_rx,
));
}
None => {
if let Err(err) = crate::twin_tls::record_replication_port(&replication_root, None) {
let _ = bound_tx.send(Err(err));
return Ok(());
}
}
}
if bound_tx.send(Ok((addr, authenticated))).is_err() {
return Ok(());
}
if wg_only {
gunnar_grpc::router_with_trust(ctx.clone(), grpc_request_id, grpc_trust.as_ref())
.context("configuring mutual TLS on the gRPC control plane")?;
let (control_stop_tx, control_stop_rx) = tokio::sync::watch::channel(false);
let serve_control = move |l: tokio::net::TcpListener, mut stop: tokio::sync::watch::Receiver<bool>| {
let router = gunnar_grpc::router_with_trust(ctx.clone(), grpc_request_id, grpc_trust.as_ref());
async move {
let router = match router {
Ok(r) => r,
Err(err) => {
tracing::error!(%err, "the control plane's tunnel door could not be configured");
return;
}
};
let at = l.local_addr().map(|a| a.to_string()).unwrap_or_default();
if let Err(err) = router
.serve_with_incoming_shutdown(tokio_stream::wrappers::TcpListenerStream::new(l), async move {
let _ = stop.wait_for(|s| *s).await;
})
.await
{
tracing::error!(%err, %at, "a control plane door stopped");
}
}
};
let doors = tokio::spawn(control_doors(listener, addr, true, crate::tunnel_addr::subscribe(), control_stop_rx, serve_control));
let _ = rx.await;
let _ = replication_stop_tx.send(true);
let _ = control_stop_tx.send(true);
let _ = doors.await;
Ok(())
} else {
gunnar_grpc::router_with_trust(ctx, grpc_request_id, grpc_trust.as_ref())
.context("configuring mutual TLS on the gRPC control plane")?
.serve_with_incoming_shutdown(
tokio_stream::wrappers::TcpListenerStream::new(listener),
async {
let _ = rx.await;
let _ = replication_stop_tx.send(true);
},
)
.await
.context("the gRPC control plane stopped")
}
});
let (addr, grpc_authenticated) = match bound_rx.await {
Ok(Ok(pair)) => pair,
Ok(Err(err)) => {
control_runtime.shutdown_background();
return Err(err);
}
Err(_) => {
control_runtime.shutdown_background();
bail!(
"the gRPC control plane task ended before it reported an address; the \
listener on {grpc_bind} never came up"
);
}
};
tracing::info!(
%addr,
control_plane_workers = workers,
control_connection_workers = CONTROL_CONNECTION_WORKERS,
"gunnar grpc control plane listening on its own runtime (a control-plane \
stall cannot reach the git listener, and its calls run on a handler runtime \
that cannot hold the connections)"
);
let mut held = ControlRuntime::new(control_runtime);
held.hold(&mut handler_runtime);
(Some(join), Some(tx), Some((addr, grpc_authenticated)), held)
};
gunnar_common::logging::spawn_signal_control();
let heartbeat = config.metrics_interval.map(|interval| {
let metrics = Arc::clone(&metrics);
tokio::spawn(async move { heartbeat(metrics, interval).await })
});
let runner = match config.runner.clone() {
None => None,
Some(runner_config) => {
if let Err(err) = crate::sandbox::available() {
if !runner_config.allow_network {
bail!(
"the runner was asked for hermetic builds and this kernel cannot \
give it a sandbox: {}\n\n\
Refusing to start rather than building with the network up and \
signing records that say otherwise. Pass --runner-allow-network to \
build anyway; every record will say the network was up.",
err.explain()
);
}
}
let stores = Arc::clone(&stores);
let bus = Arc::clone(&bus);
tracing::info!(
workspace = %runner_config.workspace.display(),
builder = %runner_config.builder,
hermetic = !runner_config.allow_network,
"the build runner is watching this server's pushes"
);
Some(std::thread::spawn(move || {
let end = crate::runner::watch(&stores, &bus, &runner_config, |outcome| {
match outcome {
crate::runner::Watched::Built(built) => tracing::info!(
target: "gunnar::runner",
event = "runner.built",
repo = %built.record.source_repo,
commit = %built.record.source_commit,
recipe = %built.record.recipe,
artifacts = built.record.subjects.len(),
"a push was built and attested"
),
crate::runner::Watched::Failed { repo, commit, why } => {
tracing::error!(
target: "gunnar::runner",
event = "runner.failed",
%repo, %commit, %why,
"a push could not be built; the runner is still watching"
)
}
crate::runner::Watched::Skipped { repo, r#ref, why } => {
tracing::debug!(
target: "gunnar::runner",
event = "runner.skipped",
%repo, r#ref = %r#ref, %why,
"a ref moved and was not built"
)
}
}
});
match crate::runner::said_end(&end) {
(tracing::Level::ERROR, words) => {
tracing::error!(target: "gunnar::runner", event = "runner.stopped", "{words}")
}
(_, words) => {
tracing::warn!(target: "gunnar::runner", event = "runner.stopped", "{words}")
}
}
}))
}
};
let _runner = runner;
#[cfg(feature = "replication")]
{
use crate::mirror::ReplicationConfig;
let replication_config = match config.replication.clone() {
Some(explicit) => explicit,
None => ReplicationConfig::load(stores.root())
.context("reading the replication configuration")?,
};
replication
.apply(replication_config)
.context("starting replication")?;
}
#[cfg(feature = "replication")]
let standby_watch = config.standby_watch.map(|watch| {
tracing::info!(
target: "gunnar::replication",
primary = %watch.primary,
probe_interval_ms = watch.probe_interval.as_millis() as u64,
promote_after = watch.promote_after,
earliest_takeover_ms = watch.earliest_takeover().as_millis() as u64,
"watching the primary; this box takes over by itself if it stops answering"
);
let slot = Arc::clone(&replication);
tokio::spawn(async move { crate::standby_watch::watch(slot, watch).await })
});
#[cfg(feature = "replication")]
let repair = match (config.repair, grpc_endpoint) {
(Some(cfg), Some((bound, _))) => {
let own = crate::repair::own_addr(bound);
let mut entitlements = crate::repair::Entitlements::of(Arc::clone(&stores));
if let Some(control) = control_for_repair.clone() {
entitlements = entitlements.with_control(control);
}
#[cfg(feature = "metered")]
let mut entitlements = entitlements.with_slot(entitlement.clone());
if let Some(grow) = grow_for_repair.clone() {
entitlements =
entitlements.with_grow(crate::paid_growth::PaidGrowth::new(grow));
}
#[cfg(feature = "fleet-pull")]
let postman: Option<Arc<dyn crate::postman::Postman>> =
match crate::postman::Smtp::from_env() {
Ok(Some(smtp)) => Some(Arc::new(smtp)),
Ok(None) => {
let root = stores.root().to_path_buf();
Some(Arc::new(crate::postman::FedRelay {
fed: move || {
crate::mail_relay::over(crate::feed::Fed::current(), &root)
},
}))
}
Err(why) => {
tracing::warn!(target: crate::repair::TARGET, "no mail relay: a sign_in_again order is refused by name ({why})");
None
}
};
let entitlements = entitlements.with_twin(cfg.twin.clone());
#[cfg(feature = "fleet-pull")]
let entitlements = entitlements.with_postman(postman);
let stores = Arc::clone(&stores);
crate::repair::arm(cfg.clone(), own.clone(), tokio::runtime::Handle::current());
Some(tokio::spawn(async move {
crate::repair::boot(Arc::clone(&stores), cfg.clone(), own.clone()).await;
#[cfg(not(feature = "fleet-pull"))]
let _ = (entitlements, own);
#[cfg(feature = "fleet-pull")]
if let Some(fleet) =
crate::repair::fleet_when_fed(cfg.fleet.clone(), crate::repair::FED_EVERY)
.await
{
let estate = crate::repair::ServerEstate::new(stores.root(), &cfg, own.clone())
.map_err(|why| tracing::warn!(target: crate::repair::TARGET, "the pull channel runs without a repair key: every order is refused by name ({why})"))
.ok();
let (_keep, stop) = tokio::sync::watch::channel(false);
crate::repair::pull_loop(stores, fleet, estate, entitlements, own, stop)
.await;
}
}))
}
(Some(_), None) => {
tracing::warn!(target: "gunnar::repair", "the boot check and the pull channel need the control plane, which is off");
None
}
(None, _) => None,
};
#[cfg(feature = "webhooks")]
{
let webhook_config = crate::webhook::WebhookConfig::load(stores.root())
.context("reading the webhook configuration")?;
webhook
.apply(webhook_config)
.context("starting outbound webhook delivery")?;
}
#[cfg(feature = "replication")]
for view in [Some(config.replication_view.clone()), control_view.clone()]
.into_iter()
.flatten()
{
let slot = Arc::clone(&replication);
view.fill(move || {
slot.replicator()
.as_deref()
.map(|replicator| {
let mut report = crate::replication_view::report(replicator);
report.refs = replicator.ref_coverage_cached();
report
})
.unwrap_or_default()
});
}
let _ = &control_view;
Ok(Server {
ssh: Some(ssh),
grpc,
grpc_shutdown,
grpc_endpoint,
control_runtime,
stores,
bootstrap_window_open,
host_key_fingerprint,
metrics,
heartbeat,
#[cfg(feature = "replication")]
standby_watch,
#[cfg(feature = "replication")]
repair,
bus,
#[cfg(feature = "http")]
http,
scrape,
#[cfg(feature = "replication")]
replication,
#[cfg(feature = "webhooks")]
webhook,
})
}
pub fn bus(&self) -> &Arc<EventBus> {
&self.bus
}
#[cfg(feature = "replication")]
pub fn replication(&self) -> Option<Arc<crate::mirror::Replicator>> {
self.replication.replicator()
}
#[cfg(feature = "replication")]
pub fn replication_slot(&self) -> &Arc<crate::mirror::ReplicationSlot> {
&self.replication
}
#[cfg(feature = "webhooks")]
pub fn webhook_slot(&self) -> &Arc<crate::webhook::WebhookSlot> {
&self.webhook
}
pub fn host_key_fingerprint(&self) -> &str {
&self.host_key_fingerprint
}
pub fn ssh_addr(&self) -> SocketAddr {
self.ssh().local_addr()
}
pub fn metrics_addr(&self) -> Option<SocketAddr> {
self.scrape.as_ref().map(|(_, addr)| *addr)
}
#[cfg(feature = "http")]
pub fn http_addr(&self) -> Option<SocketAddr> {
self.http.as_ref().map(|(_, addr, _)| *addr)
}
#[cfg(feature = "http")]
pub fn http_tls(&self) -> bool {
self.http.as_ref().is_some_and(|(_, _, tls)| *tls)
}
fn ssh(&self) -> &BoundServer {
self.ssh
.as_ref()
.expect("the SSH listener is taken only by shutdown, which consumes the Server")
}
pub fn grpc_addr(&self) -> Option<SocketAddr> {
self.grpc_endpoint.map(|(addr, _)| addr)
}
pub fn grpc_endpoint(&self) -> Option<(SocketAddr, bool)> {
self.grpc_endpoint
}
pub fn bootstrap_window_open(&self) -> bool {
self.bootstrap_window_open
}
pub fn stores(&self) -> &Arc<ServerStores> {
&self.stores
}
pub async fn wait(self) -> Result<()> {
let signal = appliance_pid1::runtime::stop_request().await;
tracing::info!(signal, "shutting both listeners down");
self.shutdown().await
}
pub async fn shutdown(mut self) -> Result<()> {
if let Some(beat) = self.heartbeat.take() {
beat.abort();
}
#[cfg(feature = "replication")]
if let Some(watch) = self.standby_watch.take() {
watch.abort();
}
#[cfg(feature = "replication")]
if let Some(repair) = self.repair.take() {
repair.abort();
}
#[cfg(feature = "replication")]
self.replication.stop();
#[cfg(feature = "webhooks")]
self.webhook.stop();
self.log_metrics();
if let Some(tx) = self.grpc_shutdown.take() {
let _ = tx.send(());
}
if let Some(join) = self.grpc.take() {
let abort = join.abort_handle();
match tokio::time::timeout(GRPC_SHUTDOWN_GRACE, join).await {
Ok(Ok(result)) => result?,
Ok(Err(err)) if err.is_cancelled() => {}
Ok(Err(err)) => return Err(anyhow::anyhow!("the gRPC task panicked: {err}")),
Err(_elapsed) => {
tracing::warn!(
grace = ?GRPC_SHUTDOWN_GRACE,
"the gRPC control plane did not stop gracefully; almost certainly an \
open Events.Watch stream, which by design never ends. Aborting it so \
the process can exit."
);
abort.abort();
}
}
}
self.control_runtime.shutdown();
if let Some(ssh) = self.ssh.take() {
ssh.shutdown("gunnar is shutting down")
.await
.context("stopping the SSH listener")?;
}
Ok(())
}
pub fn admission(&self) -> gunnar_ssh::AdmissionStats {
self.ssh().admission()
}
pub fn metrics(&self) -> MetricsSnapshot {
self.metrics.snapshot()
}
pub fn metrics_handle(&self) -> &Arc<ServerMetrics> {
&self.metrics
}
pub fn log_metrics(&self) {
emit_metrics(&self.metrics.snapshot());
}
}
impl Drop for Server {
fn drop(&mut self) {
if let Some(beat) = self.heartbeat.take() {
beat.abort();
}
#[cfg(feature = "replication")]
if let Some(watch) = self.standby_watch.take() {
watch.abort();
}
#[cfg(feature = "replication")]
if let Some(repair) = self.repair.take() {
repair.abort();
}
#[cfg(feature = "replication")]
self.replication.stop();
#[cfg(feature = "webhooks")]
self.webhook.stop();
self.control_runtime.shutdown();
}
}
pub fn announce_readiness(
readiness: &Readiness,
data_root: &std::path::Path,
grpc_bind: SocketAddr,
no_grpc: bool,
) -> String {
match readiness {
Readiness::Ready => {
let line = format!("identity table ready at {}", data_root.display());
tracing::info!(data_root = %data_root.display(), "{line}");
line
}
Readiness::AwaitingBootstrap if no_grpc => {
let line = format!(
"NO IDENTITY HAS EVER BEEN ENROLLED at {}, and the gRPC control plane is \
DISABLED, so there is no way to enrol one. Nobody can reach this server. Start \
without --no-grpc and call Bootstrap.Enroll, or pass --admin-key.",
data_root.display()
);
tracing::error!(data_root = %data_root.display(), "{line}");
line
}
Readiness::AwaitingBootstrap => {
let line = format!(
"NO IDENTITY HAS EVER BEEN ENROLLED at {}. Bootstrap.Enroll on {grpc_bind} is \
open to ANY caller that can reach it, with NO credential, and it creates a \
gunnar-admin. It closes permanently on the first success and deleting \
identities does not reopen it. Enrol now, or bind the control plane where \
nobody else can reach it.",
data_root.display()
);
tracing::warn!(
data_root = %data_root.display(),
%grpc_bind,
bootstrap_window = "OPEN",
"{line}"
);
line
}
Readiness::SealedAndEmpty { record } => {
let who = match record {
Some(record) => format!(
"was bootstrapped by \"{}\" at {}",
record.principal_id, record.at
),
None => "holds principals but no keys and carries no bootstrap record, so it \
was written before that record existed, and"
.to_owned(),
};
let line = format!(
"this server {who} its identity table at {} now holds no keys. NOBODY CAN \
AUTHENTICATE against it, and the enrolment window stays shut — deleting \
identities does not reopen it. Add a key with `gunnar key add` while the \
server is down.",
data_root.display()
);
tracing::error!(
data_root = %data_root.display(),
bootstrap_window = "SHUT",
"{line}"
);
line
}
}
}
fn bind_grpc_listener(addr: SocketAddr, device: Option<&str>) -> Result<tokio::net::TcpListener> {
use socket2::{Domain, Protocol, Socket, Type};
let domain = if addr.is_ipv6() {
Domain::IPV6
} else {
Domain::IPV4
};
let sock = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))
.context("opening the gRPC listener socket")?;
sock.set_reuse_address(true)
.context("SO_REUSEADDR on the gRPC listener")?;
if let Some(dev) = device {
sock.bind_device(Some(dev.as_bytes()))
.with_context(|| format!("SO_BINDTODEVICE {dev} on the gRPC listener"))?;
}
sock.bind(&addr.into())
.with_context(|| format!("bind {addr}"))?;
sock.listen(1024).context("listen on the gRPC listener")?;
sock.set_nonblocking(true)
.context("non-blocking gRPC listener")?;
tokio::net::TcpListener::from_std(sock.into())
.context("registering the gRPC listener with the runtime")
}
pub const GRPC_BIND_PATIENCE: std::time::Duration = std::time::Duration::from_secs(30);
pub const GRPC_BIND_RETRY_EVERY: std::time::Duration = std::time::Duration::from_millis(500);
fn bind_failed_because_the_address_is_not_here_yet(err: &anyhow::Error) -> bool {
err.chain()
.filter_map(|cause| cause.downcast_ref::<std::io::Error>())
.any(|io| io.kind() == std::io::ErrorKind::AddrNotAvailable)
}
async fn bind_when_the_address_arrives<T, F>(
mut attempt: F,
patience: std::time::Duration,
every: std::time::Duration,
) -> (Result<T>, u32)
where
F: FnMut() -> Result<T>,
{
let started = std::time::Instant::now();
let mut retries = 0u32;
loop {
match attempt() {
Ok(bound) => {
if retries > 0 {
tracing::info!(
retries,
waited_ms = started.elapsed().as_millis() as u64,
"the gRPC bind address arrived; the control plane is bound"
);
}
return (Ok(bound), retries);
}
Err(err) if bind_failed_because_the_address_is_not_here_yet(&err) => {
if started.elapsed() >= patience {
return (
Err(err.context(format!(
"the address was on no interface for {} s ({retries} retries)",
patience.as_secs()
))),
retries,
);
}
if retries == 0 {
tracing::warn!(
patience_s = patience.as_secs(),
"the gRPC bind address is on no interface yet (EADDRNOTAVAIL); waiting for \
it. A tailnet or VPN interface that comes up after this service is the \
usual cause"
);
}
retries += 1;
tokio::time::sleep(every).await;
}
Err(err) => return (Err(err), retries),
}
}
}
#[cfg(test)]
mod grpc_bind_patience_tests {
use super::*;
use std::time::Duration;
fn not_here_yet() -> anyhow::Error {
anyhow::Error::from(std::io::Error::from(std::io::ErrorKind::AddrNotAvailable))
.context("bind 100.64.0.1:50061")
}
#[test]
fn an_address_that_arrives_later_is_bound_and_the_retries_are_counted() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let mut left = 2;
let (bound, retries) = rt.block_on(bind_when_the_address_arrives(
|| {
if left > 0 {
left -= 1;
Err(not_here_yet())
} else {
Ok(7u8)
}
},
Duration::from_secs(5),
Duration::from_millis(1),
));
assert_eq!(bound.unwrap(), 7);
assert_eq!(retries, 2);
}
#[test]
fn an_address_in_use_is_not_waited_for() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let mut calls = 0;
let (bound, retries) = rt.block_on(bind_when_the_address_arrives(
|| -> Result<u8> {
calls += 1;
Err(anyhow::Error::from(std::io::Error::from(
std::io::ErrorKind::AddrInUse,
)))
},
Duration::from_secs(5),
Duration::from_millis(1),
));
assert!(bound.is_err());
assert_eq!(retries, 0);
assert_eq!(calls, 1);
}
#[test]
fn an_address_that_never_comes_gives_up_after_the_patience() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let (bound, retries) = rt.block_on(bind_when_the_address_arrives(
|| -> Result<u8> { Err(not_here_yet()) },
Duration::from_millis(30),
Duration::from_millis(5),
));
let msg = format!("{:#}", bound.unwrap_err());
assert!(msg.contains("on no interface for 0 s"), "{msg}");
assert!(retries >= 3, "retries={retries}");
}
#[test]
fn the_real_bind_error_for_an_absent_address_is_classified_as_not_here_yet() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_io()
.build()
.unwrap();
let _guard = rt.enter();
let err = bind_grpc_listener("192.0.2.1:0".parse().unwrap(), None)
.expect_err("TEST-NET-1 is on no interface");
assert!(
bind_failed_because_the_address_is_not_here_yet(&err),
"{err:#}"
);
let in_use = anyhow::Error::from(std::io::Error::from(std::io::ErrorKind::AddrInUse));
assert!(!bind_failed_because_the_address_is_not_here_yet(&in_use));
}
}
#[cfg(test)]
mod grpc_bind_device_tests {
use super::*;
#[test]
fn a_device_the_machine_does_not_have_is_refused_by_name() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_io()
.build()
.unwrap();
let _guard = rt.enter();
let err = bind_grpc_listener("127.0.0.1:0".parse().unwrap(), Some("nosuchdev0"))
.expect_err("a nonexistent device cannot be bound");
let msg = format!("{err:#}");
assert!(msg.contains("SO_BINDTODEVICE nosuchdev0"), "{msg}");
}
#[test]
fn an_address_binds_and_so_does_the_loopback_device() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_io()
.build()
.unwrap();
let _guard = rt.enter();
let plain = bind_grpc_listener("127.0.0.1:0".parse().unwrap(), None).unwrap();
assert!(plain.local_addr().unwrap().port() != 0);
match bind_grpc_listener("127.0.0.1:0".parse().unwrap(), Some("lo")) {
Ok(l) => assert!(l.local_addr().unwrap().port() != 0),
Err(e) => assert!(format!("{e:#}").contains("SO_BINDTODEVICE lo"), "{e:#}"),
}
}
}
fn resolve_grpc_bind(spec: &BindSpec) -> Result<Option<SocketAddr>> {
match spec
.resolve_with(|name| {
appliance_pid1::net::interface_ipv4(name).map_err(|e| format!("{e:#}"))
})
.map_err(|why| anyhow::anyhow!("{why}"))?
{
gunnar_common::bindspec::Resolved::Bound(addr) => Ok(Some(addr)),
gunnar_common::bindspec::Resolved::NoAddress { interface, why } => {
tracing::error!(
interface = %interface,
"{why} The gRPC control plane will NOT start. The git listener is unaffected \
and this server still serves and authenticates pushes and fetches over SSH."
);
Ok(None)
}
}
}
fn read_key_argument(spec: &str) -> Result<String> {
match spec.strip_prefix('@') {
Some(path) => std::fs::read_to_string(path)
.with_context(|| format!("reading the public key from {path}"))
.map(|text| text.trim().to_owned()),
None => Ok(spec.trim().to_owned()),
}
}
pub(crate) fn host_key_banner_line(fingerprint: &str) -> String {
format!("sshkey {fingerprint} (pin THIS; a key scanned off the port proves nothing)")
}
fn russh_private_key(openssh: &[u8]) -> Result<gunnar_ssh::russh::keys::PrivateKey> {
gunnar_ssh::russh::keys::PrivateKey::from_openssh(openssh)
.map_err(|err| anyhow::anyhow!("the stored host key did not parse: {err}"))
}
pub async fn run(stores: Arc<ServerStores>, config: ServeConfig, check_only: bool) -> Result<()> {
let _ = crate::limits::nofile();
use crate::narrate::Stage;
crate::narrate::install();
crate::narrate::data_home(stores.root());
crate::narrate::at(
Stage::Config,
&format!(
"data root {}, engine {:?}, ssh {}, grpc {:?}, twin_role={}, is_twin={}",
stores.root().display(),
stores.engine(),
config.ssh_bind,
config.grpc_bind,
config.twin_role,
stores.is_twin(),
),
);
crate::narrate::at(
Stage::Network,
"bringing the declared interfaces up over netlink",
);
let dhcp_session = crate::boot::bring_up_network()?;
crate::narrate::at(
Stage::Lease,
&format!("{} DHCP lease(s) held", dhcp_session.len()),
);
crate::narrate::at(
Stage::Sink,
"the boot log is kept for the installer (its bootstrap listener serves it)",
);
crate::clock::synchronise_at_boot().await;
crate::clock::arm_triggers();
let mut config = config;
config.grpc_bind = crate::boot::role_bound_here(config.grpc_bind)
.context("resolving the --grpc-bind role to this box's NIC")?;
if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
tracing::info!(interface = %interface, "the control plane's utility door is {interface} on this box");
}
if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
let has_address = |i: &str| matches!(appliance_pid1::net::interface_ipv4(i), Ok(Some(_)));
if !has_address(interface) && appliance_pid1::net::dhcp_pending(interface) {
tracing::info!(interface = %interface, "the control plane waits for {interface}'s address (PID 1 asks DHCP again in the background)");
while !has_address(interface) && appliance_pid1::net::dhcp_pending(interface) {
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
}
}
}
let resolved = resolve_grpc_bind(&config.grpc_bind).with_context(|| {
"resolving --grpc-bind after the network came up. A medium that names a NIC this \
machine does not have is a declaration that lies, and gunnar stops rather than \
falling back to a wildcard bind: an unauthenticated control plane on every \
interface is the defect the interface spelling exists to close."
})?;
let resolved_grpc_bind = match resolved {
Some(addr) => addr,
None => {
config.no_grpc = true;
gunnar_grpc::DEFAULT_BIND
}
};
if let BindSpec::Interface { interface, .. } = &config.grpc_bind {
config.grpc_bind_device = Some(interface.clone());
}
config.grpc_bind = BindSpec::Addr(resolved_grpc_bind);
if !config.no_grpc && !resolved_grpc_bind.ip().is_loopback() {
tracing::info!(
grpc_bind = %resolved_grpc_bind,
"gRPC control plane will bind a NON-loopback address"
);
}
crate::boot::announce_install_mode(stores.root());
crate::disktest::spawn(stores.root().to_path_buf());
if appliance_pid1::net::is_init() {
tracing::info!(
"running as pid 1: no init system underneath, so this process handles SIGTERM and \
reaps re-parented orphans itself"
);
}
let reaper = tokio::spawn(appliance_pid1::runtime::reaper());
let tombstone_sweep = crate::stores::spawn_tombstone_sweep(Arc::clone(&stores));
let flow_band: Arc<std::sync::Mutex<Option<gunnar_core::Fleet>>> =
Arc::new(std::sync::Mutex::new(None));
let flow_flush = crate::stores::spawn_flow_flush(Arc::clone(&stores), Arc::clone(&flow_band));
config
.flow_view
.fill(move || *flow_band.lock().expect("the flow band"));
let flow_at_exit = Arc::clone(stores.flow());
let replication_view = config.replication_view.clone();
crate::dashboard::spawn(crate::dashboard::Facts {
ssh_bind: config.ssh_bind,
grpc_bind: (!config.no_grpc).then_some(resolved_grpc_bind),
data_root: stores.root().to_path_buf(),
build: gunnar_common::BUILD_REF.to_owned(),
engine: Some(stores.engine()),
replication: replication_view.clone(),
tenants: config.tenants_view.clone(),
flow: config.flow_view.clone(),
});
#[cfg(feature = "dhcp-client")]
let renewals: Vec<tokio::task::JoinHandle<()>> = dhcp_session
.into_iter()
.map(|session| tokio::spawn(crate::boot::dhcp_renewal(session)))
.collect();
#[cfg(not(feature = "dhcp-client"))]
let renewals: Vec<tokio::task::JoinHandle<()>> = dhcp_session
.into_iter()
.map(|session| match session {})
.collect();
#[cfg(feature = "wg-client")]
let tunnel = crate::wg::bring_up(stores.root())?;
#[cfg(feature = "wg-client")]
let handshake = tunnel.clone().map(|t| {
tokio::spawn(crate::wg::handshake_watch(
t,
20,
std::time::Duration::from_secs(3),
))
});
#[cfg(not(feature = "wg-client"))]
let handshake: Option<tokio::task::JoinHandle<()>> = None;
let config_twin_role = config.twin_role;
let stores_is_twin = stores.is_twin();
let server = Server::start(stores, config).await?;
let staleness_watch = crate::replication_view::spawn_staleness_watch(
replication_view.clone(),
Arc::clone(server.bus()),
);
let mut banner = String::new();
{
use std::fmt::Write as _;
writeln!(banner, "ssh {}", server.ssh_addr()).ok();
writeln!(
banner,
"{}",
host_key_banner_line(server.host_key_fingerprint())
)
.ok();
match server.grpc_endpoint() {
Some((addr, _)) if server.bootstrap_window_open() => writeln!(
banner,
"grpc {addr} (NOT ENROLLED — Bootstrap.Enroll is open to anyone who can reach \
this port. `gunnar control bootstrap <name>` claims it, once.)"
)
.ok(),
Some((addr, true)) => writeln!(
banner,
"grpc {addr} (authenticated: sshsig + the ACL model)"
)
.ok(),
Some((addr, false)) => writeln!(
banner,
"grpc {addr} (UNAUTHENTICATED — the escape hatch is on)"
)
.ok(),
None => writeln!(banner, "grpc disabled").ok(),
};
#[cfg(feature = "http")]
if let Some(addr) = server.http_addr() {
if server.http_tls() {
writeln!(
banner,
"https {addr} (TLS terminated here; clients pin the issuer)"
)
.ok();
} else {
writeln!(banner, "http {addr}").ok();
}
}
if let Some(addr) = server.metrics_addr() {
writeln!(
banner,
"metrics http://{addr}{} (no authentication{})",
crate::scrape::SCRAPE_PATH,
if addr.ip().is_loopback() {
""
} else {
"; NOT on loopback"
}
)
.ok();
}
#[cfg(feature = "wg-client")]
if let Some(tunnel) = &tunnel {
writeln!(banner, "{}", tunnel.banner_line()).ok();
}
writeln!(
banner,
"log {} (SIGUSR1 → one rung more verbose, SIGUSR2 → back; pid {})",
gunnar_common::logging::current().unwrap_or_else(|| "?".to_owned()),
std::process::id()
)
.ok();
}
print!("{banner}");
appliance_pid1::console::mirror_to_gfx_console(&banner);
crate::narrate::at(Stage::Listener, &format!("ssh BOUND {}", server.ssh_addr()));
match server.grpc_endpoint() {
Some((addr, _)) => crate::narrate::at(
Stage::Listener,
&format!(
"grpc BOUND {addr}{}",
if server.bootstrap_window_open() {
" (NOT ENROLLED — Bootstrap.Enroll is open)"
} else {
""
}
),
),
None => crate::narrate::at(Stage::Listener, "grpc NOT bound — the control plane is off"),
}
#[cfg(feature = "http")]
if let Some(addr) = server.http_addr() {
crate::narrate::at(
Stage::Listener,
&format!(
"{} BOUND {addr}",
if server.http_tls() { "https" } else { "http" }
),
);
}
if let Some(addr) = server.metrics_addr() {
crate::narrate::at(Stage::Listener, &format!("metrics BOUND {addr}"));
}
crate::narrate::at(
Stage::Role,
&format!(
"{} (declared twin_role={}, data home says is_twin={})",
if stores_is_twin || config_twin_role {
"TWIN"
} else {
"PRIMARY"
},
config_twin_role,
stores_is_twin,
),
);
if check_only {
reaper.abort();
tombstone_sweep.abort();
flow_flush.abort();
staleness_watch.abort();
for renewal in &renewals {
renewal.abort();
}
if let Some(handshake) = &handshake {
handshake.abort();
}
server.shutdown().await?;
return Ok(());
}
crate::narrate::at(
Stage::Serving,
&format!("up at {} s", crate::narrate::uptime_s(),),
);
let result = server.wait().await;
reaper.abort();
tombstone_sweep.abort();
if let Err(err) = flow_at_exit.flush(gunnar_core::now_ms()) {
tracing::warn!(
target: "gunnar::store",
event = "flow.flush_failed",
%err,
"the flow ledger could not be written on shutdown; up to one flush interval of \
today's activity counters is lost"
);
}
flow_flush.abort();
staleness_watch.abort();
for renewal in &renewals {
renewal.abort();
}
if let Some(handshake) = &handshake {
handshake.abort();
}
result
}
#[cfg(test)]
mod tests {
use super::read_key_argument;
use super::*;
#[test]
fn the_banner_names_the_host_key_in_full() {
const FP: &str = "SHA256:z/fSv0Z0RZS+Lccbc6ZoOWwt/fbj1VFJGSBKQbo4icE";
let line = host_key_banner_line(FP);
assert!(line.contains(FP), "{line}");
assert!(line.starts_with("sshkey "), "one searchable token: {line}");
assert!(
!line.contains('\n'),
"one line, or the banner is not a banner: {line}"
);
assert!(
line.contains("scanned off the port proves nothing"),
"{line}"
);
}
#[test]
fn the_blocking_pool_is_larger_than_the_transfer_limit() {
let transfers = gunnar_ssh::SshConfig::default()
.max_concurrent_transfers
.expect("the appliance default must have a transfer limit");
assert!(
transfers < GIT_BLOCKING_THREADS,
"max_concurrent_transfers ({transfers}) is not below GIT_BLOCKING_THREADS \
({GIT_BLOCKING_THREADS}), so the blocking pool becomes the real ceiling again and \
the admission counters would show a server that is never full while every transfer \
queues"
);
assert!(
GIT_BLOCKING_THREADS >= transfers * 2,
"the blocking pool ({GIT_BLOCKING_THREADS}) leaves no headroom over \
{transfers} concurrent transfers for the other blocking work a transfer needs: \
pack indexing, ref transactions, the arrow log"
);
}
#[test]
fn the_control_plane_gets_its_own_bounded_pool() {
let workers = control_plane_workers();
assert!(
(4..=8).contains(&workers),
"control_plane_workers() returned {workers}; below four, two slow synchronous calls \
hold every worker and the whole administrative surface queues behind them, and an \
unbounded pool recreates the starved listener with extra steps"
);
}
#[test]
fn a_literal_key_and_an_at_path_read_the_same() {
let dir = tempfile::tempdir().expect("a temp dir");
let path = dir.path().join("id.pub");
std::fs::write(&path, "ssh-ed25519 AAAAC3Nz literal\n").expect("write");
assert_eq!(
read_key_argument("ssh-ed25519 AAAAC3Nz literal").expect("literal"),
read_key_argument(&format!("@{}", path.display())).expect("from file"),
"`--admin-key @file` and `--admin-key '<line>'` must produce the same key, or a \
bootstrap that works one way fails the other"
);
}
#[test]
fn a_quota_far_narrower_than_the_affinity_is_reported_and_a_matching_one_is_not() {
assert_eq!(
spread_budget(Some(32), Some(8.0)),
Some((32, 8)),
"eight CPUs of quota spread over thirty-two is the condition that cost 2.7x the \
CPU per request at 32 clients; it must be said out loud"
);
assert_eq!(spread_budget(Some(8), Some(8.0)), None);
assert_eq!(spread_budget(Some(16), Some(8.0)), Some((16, 8)));
assert_eq!(spread_budget(Some(15), Some(8.0)), None);
assert_eq!(spread_budget(Some(32), Some(1.5)), Some((32, 2)));
assert_eq!(spread_budget(Some(32), None), None);
assert_eq!(spread_budget(None, Some(8.0)), None);
}
#[test]
fn a_cpu_list_is_counted_whether_it_is_ranges_singletons_or_both() {
assert_eq!(count_cpu_list("0-31"), Some(32));
assert_eq!(count_cpu_list("0-7"), Some(8));
assert_eq!(
count_cpu_list("3"),
Some(1),
"one pinned CPU is one CPU, not none"
);
assert_eq!(count_cpu_list("0,2,4,6"), Some(4));
assert_eq!(count_cpu_list("0-3,8,10-11"), Some(7));
assert_eq!(
count_cpu_list(" 0-3 , 8 \n"),
Some(5),
"the kernel line is padded"
);
assert_eq!(count_cpu_list(""), None);
assert_eq!(count_cpu_list("banana"), None);
assert_eq!(
count_cpu_list("7-3"),
None,
"a reversed range is not 5 CPUs"
);
}
#[test]
fn the_affinity_of_this_process_is_read_as_a_count_of_cpus() {
let n = cpus_allowed().expect("Linux always publishes Cpus_allowed_list");
assert!(
(1..=1 << 20).contains(&n),
"this process reports {n} allowed CPUs, which is not a count of CPUs"
);
let online = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
assert!(
n >= online,
"the affinity ({n}) is narrower than what std reports as available ({online}), \
which would make every comparison against it too quiet"
);
}
#[test]
fn a_missing_key_file_is_an_error_not_an_empty_key() {
let err = read_key_argument("@/nonexistent/gunnar/key.pub")
.expect_err("a missing key file must fail");
assert!(format!("{err:#}").contains("reading the public key"));
}
}
pub fn front_control(
stores: &crate::stores::ServerStores,
anonymous: gunnar_ssh::AnonymousPolicy,
disk: Arc<gunnar_ssh::DiskHeadroomGate>,
#[cfg(feature = "metered")] entitlement: Option<&Arc<crate::entitlement::EntitlementSlot>>,
) -> gunnar_ssh::Control {
#[allow(unused_mut)]
stores.install_space_ledger(Arc::clone(disk.ledger()));
let control = stores
.ssh_control()
.with_anonymous_policy(anonymous)
.with_plan_gate(disk as Arc<dyn gunnar_ssh::PlanGate>);
#[cfg(feature = "metered")]
let control = match entitlement {
Some(slot) => control.with_plan_gate(Arc::clone(slot) as Arc<dyn gunnar_ssh::PlanGate>),
None => control,
};
stores.install_plan_gates(control.plan_gates().to_vec());
control
}
pub fn record_replication_door(
root: &std::path::Path,
utility: SocketAddr,
tunnel: Option<SocketAddr>,
) -> Result<()> {
crate::twin_tls::record_replication_port(root, Some(utility.port()))?;
let hosts: Vec<std::net::IpAddr> = match tunnel {
Some(t) => [t.ip(), utility.ip()]
.into_iter()
.filter(|ip| !ip.is_unspecified())
.collect(),
None => vec![],
};
crate::twin_tls::record_replication_host(root, &hosts)
}
#[allow(clippy::too_many_arguments)]
pub async fn replication_door(
listener: tokio::net::TcpListener,
base: SocketAddr,
explicit: bool,
wg_only: bool,
ctx: gunnar_grpc::ServiceContext,
request_ids: gunnar_grpc::RequestIdSource,
trust: gunnar_grpc::Trust,
root: std::path::PathBuf,
stop: tokio::sync::watch::Receiver<bool>,
) {
let mut ctx = ctx;
ctx.config = ctx
.config
.clone()
.with_registered_client_certificate_required(true);
let serve = move |listener: tokio::net::TcpListener,
mut stop: tokio::sync::watch::Receiver<bool>| {
let router = gunnar_grpc::router_with_trust(ctx.clone(), request_ids, Some(&trust));
async move {
let router = match router {
Ok(r) => r,
Err(err) => {
tracing::error!(%err, "the mutual replication listener could not be configured; no twin can follow this box");
return;
}
};
let at = listener
.local_addr()
.map(|a| a.to_string())
.unwrap_or_default();
let served = router
.serve_with_incoming_shutdown(
tokio_stream::wrappers::TcpListenerStream::new(listener),
async move {
let _ = stop.wait_for(|s| *s).await;
},
)
.await;
if let Err(err) = served {
tracing::error!(%err, %at, "a mutual replication door stopped");
}
}
};
replication_doors_wg(
listener,
base,
explicit,
wg_only,
root,
crate::tunnel_addr::subscribe(),
stop,
serve,
)
.await
}
pub async fn replication_doors<S, F>(
listener: tokio::net::TcpListener,
base: SocketAddr,
explicit: bool,
root: std::path::PathBuf,
tunnel: tokio::sync::watch::Receiver<Option<std::net::Ipv4Addr>>,
stop: tokio::sync::watch::Receiver<bool>,
serve: S,
) where
S: Fn(tokio::net::TcpListener, tokio::sync::watch::Receiver<bool>) -> F,
F: std::future::Future<Output = ()> + Send + 'static,
{
replication_doors_wg(listener, base, explicit, false, root, tunnel, stop, serve).await
}
#[allow(clippy::too_many_arguments)]
pub async fn replication_doors_wg<S, F>(
listener: tokio::net::TcpListener,
base: SocketAddr,
explicit: bool,
wg_only: bool,
root: std::path::PathBuf,
tunnel: tokio::sync::watch::Receiver<Option<std::net::Ipv4Addr>>,
stop: tokio::sync::watch::Receiver<bool>,
serve: S,
) where
S: Fn(tokio::net::TcpListener, tokio::sync::watch::Receiver<bool>) -> F,
F: std::future::Future<Output = ()> + Send + 'static,
{
let record = move |utility: SocketAddr, door: Option<SocketAddr>| {
let at = if wg_only {
record_replication_door_wg_only(&root, utility.port(), door)
} else {
record_replication_door(&root, utility, door)
};
if let Err(err) = at {
tracing::error!(%err, "the replication doors changed but their addresses could not be recorded");
}
};
doors(
listener,
base,
wg_only,
tunnel,
stop,
serve,
move |b, t| crate::tunnel_addr::replication_extra(b, explicit, t),
record,
"mutual replication listener",
)
.await
}
pub fn record_replication_door_wg_only(
root: &std::path::Path,
port: u16,
tunnel: Option<SocketAddr>,
) -> Result<()> {
crate::twin_tls::record_replication_port(root, Some(port))?;
let hosts: Vec<std::net::IpAddr> = tunnel.map(|t| t.ip()).into_iter().collect();
crate::twin_tls::record_replication_host(root, &hosts)
}
pub async fn control_doors<S, F>(
listener: tokio::net::TcpListener,
base: SocketAddr,
wg_only: bool,
tunnel: tokio::sync::watch::Receiver<Option<std::net::Ipv4Addr>>,
stop: tokio::sync::watch::Receiver<bool>,
serve: S,
) where
S: Fn(tokio::net::TcpListener, tokio::sync::watch::Receiver<bool>) -> F,
F: std::future::Future<Output = ()> + Send + 'static,
{
doors(
listener,
base,
wg_only,
tunnel,
stop,
serve,
|_, _| None,
|_, _| {},
"control plane",
)
.await
}
#[allow(clippy::too_many_arguments)]
pub async fn doors<S, F, X, R>(
listener: tokio::net::TcpListener,
base: SocketAddr,
wg_only: bool,
mut tunnel: tokio::sync::watch::Receiver<Option<std::net::Ipv4Addr>>,
mut stop: tokio::sync::watch::Receiver<bool>,
serve: S,
extra: X,
mut record: R,
what: &'static str,
) where
S: Fn(tokio::net::TcpListener, tokio::sync::watch::Receiver<bool>) -> F,
F: std::future::Future<Output = ()> + Send + 'static,
X: Fn(SocketAddr, Option<std::net::Ipv4Addr>) -> Option<SocketAddr>,
R: FnMut(SocketAddr, Option<SocketAddr>),
{
let port = listener
.local_addr()
.map(|a| a.port())
.unwrap_or(base.port());
let base = SocketAddr::new(base.ip(), port);
let utility = if wg_only {
drop(listener);
tracing::warn!(%base, "{what}: --wg-only — the utility door is CLOSED; it answers only on the WireGuard tunnel's address");
None
} else {
Some(tokio::spawn(serve(listener, stop.clone())))
};
let mut door: Option<(SocketAddr, tokio::sync::watch::Sender<bool>)> = None;
let mut first = true;
loop {
let t = *tunnel.borrow_and_update();
let want = if wg_only {
t.map(|t| SocketAddr::new(std::net::IpAddr::V4(t), port))
} else {
extra(base, t)
};
if first || door.as_ref().map(|(a, _)| *a) != want {
first = false;
if let Some((at, close)) = door.take() {
let _ = close.send(true);
if wg_only {
tracing::warn!(%at, "{what}: the tunnel went away or moved: its door is closed, and with --wg-only nothing serves until it is back");
} else {
tracing::warn!(%at, "{what}: the tunnel went away or moved: the door on it is closed; the utility door {base} keeps serving");
}
}
if let Some(at) = want {
match tokio::net::TcpListener::bind(at).await {
Ok(l) => {
let (close, closed) = tokio::sync::watch::channel(false);
tokio::spawn(serve(l, closed));
if wg_only {
tracing::info!(%at, "{what}: up on the WireGuard tunnel's address, and only there (--wg-only)");
} else {
tracing::warn!(%at, %base, "{what} ALSO on the tunnel's address (no restart); the utility door stays");
}
door = Some((at, close));
}
Err(err) => {
tracing::error!(%err, %at, "{what}: the tunnel's door could not bind")
}
}
}
record(base, door.as_ref().map(|(a, _)| *a));
}
tokio::select! {
_ = async { let _ = stop.wait_for(|s| *s).await; } => break,
changed = tunnel.changed() => {
if changed.is_err() {
let _ = stop.wait_for(|s| *s).await;
break;
}
}
}
}
if let Some((_, close)) = door.take() {
let _ = close.send(true);
}
if let Some(u) = utility {
let _ = u.await;
}
}
#[cfg(test)]
mod replication_door_tests {
use super::*;
use std::net::Ipv4Addr;
use tokio::io::AsyncWriteExt as _;
async fn answering(
listener: tokio::net::TcpListener,
mut stop: tokio::sync::watch::Receiver<bool>,
) {
loop {
tokio::select! {
_ = async { let _ = stop.wait_for(|s| *s).await; } => return,
got = listener.accept() => if let Ok((mut c, _)) = got { let _ = c.write_all(b"R").await; },
}
}
}
async fn answers(at: SocketAddr) -> bool {
use tokio::io::AsyncReadExt as _;
let Ok(Ok(mut c)) = tokio::time::timeout(
std::time::Duration::from_secs(10),
tokio::net::TcpStream::connect(at),
)
.await
else {
return false;
};
let mut b = [0u8; 1];
matches!(
tokio::time::timeout(std::time::Duration::from_secs(10), c.read_exact(&mut b)).await,
Ok(Ok(_))
)
}
async fn settle() {
tokio::time::sleep(std::time::Duration::from_millis(150)).await;
}
async fn eventually<F, Fut>(mut cond: F) -> bool
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = bool>,
{
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(120);
loop {
if cond().await {
return true;
}
if std::time::Instant::now() >= deadline {
return false;
}
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn under_wg_only_the_utility_door_is_closed_and_only_the_tunnel_door_answers() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().to_path_buf();
let utility = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let base = utility.local_addr().unwrap();
let (tunnel_tx, tunnel_rx) = tokio::sync::watch::channel(None::<Ipv4Addr>);
let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);
let doors = tokio::spawn(replication_doors_wg(
utility,
base,
false,
true,
root.clone(),
tunnel_rx,
stop_rx,
answering,
));
let refused = |at: SocketAddr| async move {
matches!(tokio::net::TcpStream::connect(at).await, Err(e) if e.kind() == std::io::ErrorKind::ConnectionRefused)
};
assert!(
eventually(|| refused(base)).await,
"no tunnel: the utility door is closed, not served"
);
assert_eq!(crate::twin_tls::replication_host(&root), None);
let t = Ipv4Addr::new(127, 0, 0, 2);
tunnel_tx.send_replace(Some(t));
let on_tunnel = SocketAddr::new(t.into(), base.port());
assert!(
eventually(|| answers(on_tunnel)).await,
"the tunnel door answers"
);
assert!(
eventually(|| refused(base)).await,
"and a plain-IP connection is refused"
);
assert!(
eventually(|| std::future::ready(
crate::twin_tls::replication_host(&root).as_deref() == Some("127.0.0.2")
))
.await,
"the twin is told the tunnel alone: {:?}",
crate::twin_tls::replication_host(&root)
);
let control = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let cbase = control.local_addr().unwrap();
let (_ctx, crx) = tokio::sync::watch::channel(Some(t));
let (cstop_tx, cstop_rx) = tokio::sync::watch::channel(false);
let cdoors = tokio::spawn(control_doors(
control, cbase, true, crx, cstop_rx, answering,
));
assert!(
eventually(|| answers(SocketAddr::new(t.into(), cbase.port()))).await,
"the control plane answers on the tunnel"
);
assert!(
eventually(|| refused(cbase)).await,
"and refuses the plain IP"
);
stop_tx.send_replace(true);
cstop_tx.send_replace(true);
tokio::time::timeout(std::time::Duration::from_secs(5), doors)
.await
.expect("the doors stop")
.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(5), cdoors)
.await
.expect("the control doors stop")
.unwrap();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_utility_door_answers_whatever_the_tunnel_does() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().to_path_buf();
let utility = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let base = utility.local_addr().unwrap();
let (tunnel_tx, tunnel_rx) = tokio::sync::watch::channel(None::<Ipv4Addr>);
let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);
let doors = tokio::spawn(replication_doors(
utility,
base,
false,
root.clone(),
tunnel_rx,
stop_rx,
answering,
));
assert!(eventually(|| answers(base)).await);
assert!(
eventually(|| std::future::ready(
crate::twin_tls::replication_port(&root) == Some(base.port())
))
.await,
"the twin is told the utility port: {:?}",
crate::twin_tls::replication_port(&root)
);
assert_eq!(crate::twin_tls::replication_host(&root), None);
tunnel_tx.send_replace(Some(Ipv4Addr::new(10, 254, 254, 254)));
settle().await;
assert!(
eventually(|| answers(base)).await,
"a tunnel door that cannot bind must not cost the utility door"
);
assert_eq!(crate::twin_tls::replication_host(&root), None);
let t = Ipv4Addr::new(127, 0, 0, 2);
tunnel_tx.send_replace(Some(t));
let on_tunnel = SocketAddr::new(t.into(), base.port());
assert!(
eventually(|| answers(on_tunnel)).await,
"the tunnel door answers"
);
assert!(answers(base).await, "and the utility door still does");
assert!(
eventually(|| std::future::ready(
crate::twin_tls::replication_host(&root).as_deref() == Some("127.0.0.2,127.0.0.1")
))
.await,
"the twin is told tunnel first, utility next: {:?}",
crate::twin_tls::replication_host(&root)
);
tunnel_tx.send_replace(None);
assert!(
eventually(|| async move { !answers(on_tunnel).await }).await,
"the tunnel door is closed"
);
assert!(answers(base).await);
assert!(
eventually(|| std::future::ready(crate::twin_tls::replication_host(&root).is_none()))
.await,
"no host once the tunnel is gone: {:?}",
crate::twin_tls::replication_host(&root)
);
stop_tx.send_replace(true);
tokio::time::timeout(std::time::Duration::from_secs(5), doors)
.await
.expect("the doors stop")
.unwrap();
}
}