Menu
AkurAI-Build
publicLatest change a9ee1aad5871410be579b77bfd0277c1d4e891d4 - policy: enforce SDP-002..006/010 in config::parse, add fleet-audit command by Ólafur Búi Ólafsson
//! Operate the AkurAI EC2 platform box (mail.olibuijr.com).
//!
//! A faithful port of the standalone `akurai-ec2` bash CLI: same argument order,
//! same defaults, same environment variables, same stdout contract. The box runs
//! live production services, so behaviour here is deliberately conservative —
//! every verb does what the shell script did, no more.
pub mod deploy;
pub mod dns;
pub mod host;
pub mod ops;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use anyhow::{Context, Result, anyhow, bail};
use clap::Subcommand;
/// Elastic IP of the platform box. Used to locate its security group.
pub const PUBLIC_IP: &str = "3.94.46.219";
/// Inclusive lower bound of the application port pool.
pub const APP_POOL_LO: u16 = 8094;
/// Inclusive upper bound of the application port pool.
pub const APP_POOL_HI: u16 = 8199;
/// Mirrors the bash `AKURAI_EC2_SSH_RECOVERED` guard: the SSH self-heal runs at
/// most once per process.
static SSH_RECOVERED: AtomicBool = AtomicBool::new(false);
/// `say()` — human-facing progress on stdout.
pub fn say(msg: &str) {
println!("\x1b[1;34m▸ {msg}\x1b[0m");
}
/// The `die()` styling, for the paths where bash reports and keeps going.
fn cry(msg: &str) {
eprintln!("\x1b[1;31m✗ {msg}\x1b[0m");
}
/// Single-quote a value for safe interpolation into a remote shell command.
pub fn shell_quote(value: &str) -> String {
let mut quoted = String::with_capacity(value.len() + 2);
quoted.push('\'');
for ch in value.chars() {
if ch == '\'' {
quoted.push_str("'\\''");
} else {
quoted.push(ch);
}
}
quoted.push('\'');
quoted
}
/// `${VAR:-default}`: unset *or empty* falls back.
fn env_or(name: &str, fallback: impl FnOnce() -> Result<String>) -> Result<String> {
match std::env::var(name) {
Ok(value) if !value.is_empty() => Ok(value),
_ => fallback(),
}
}
fn env_present(name: &str) -> bool {
std::env::var(name).is_ok_and(|value| !value.is_empty())
}
fn home() -> Result<PathBuf> {
let home = std::env::var("HOME").context("HOME is not set")?;
if home.is_empty() {
bail!("HOME is empty");
}
Ok(PathBuf::from(home))
}
/// `command -v <name>` without spawning a process.
fn has_command(name: &str) -> bool {
let Ok(path) = std::env::var("PATH") else {
return false;
};
std::env::split_paths(&path).any(|dir| dir.join(name).is_file())
}
/// Shared connection settings resolved from the environment.
pub struct Ec2 {
pub ssh_host: String,
pub ssh_config: PathBuf,
pub public_ip: &'static str,
pub ports_registry: PathBuf,
}
impl Ec2 {
pub fn from_env() -> Result<Self> {
let ssh_host = env_or("AKURAI_EC2_SSH", || Ok("akurai-ec2".to_string()))?;
let ssh_config = env_or("AKURAI_EC2_SSH_CONFIG", || {
Ok(home()?.join(".ssh").join("config").display().to_string())
})?;
let ports_registry = env_or("AKURAI_PORTS_REGISTRY", || {
Ok(home()?
.join(".config")
.join("akurai-ec2")
.join("ports.registry")
.display()
.to_string())
})?;
Ok(Self {
ssh_host,
ssh_config: PathBuf::from(ssh_config),
public_ip: PUBLIC_IP,
ports_registry: PathBuf::from(ports_registry),
})
}
/// Run a command on the box. Returns captured stdout on success.
pub fn ssh(&self, command: &str) -> Result<String> {
let (code, stdout) = self.invoke(command, None, true)?;
if code != 0 {
bail!("ssh {}: command failed (exit {code})", self.ssh_host);
}
Ok(stdout)
}
/// Stream a command to the box, inheriting stdio (for interactive/long output).
pub fn ssh_streaming(&self, command: &str) -> Result<()> {
let code = self.ssh_streaming_status(command)?;
if code != 0 {
bail!("ssh {}: command failed (exit {code})", self.ssh_host);
}
Ok(())
}
/// Stream a command to the box and return its exit status instead of failing.
pub fn ssh_streaming_status(&self, command: &str) -> Result<i32> {
let (code, _) = self.invoke(command, None, false)?;
Ok(code)
}
/// Run a command on the box with `stdin` piped in. Returns captured stdout.
pub fn ssh_stdin(&self, command: &str, stdin: &[u8]) -> Result<String> {
let (code, stdout) = self.invoke(command, Some(stdin), true)?;
if code != 0 {
bail!("ssh {}: command failed (exit {code})", self.ssh_host);
}
Ok(stdout)
}
/// Run a command on the box with `stdin` piped in, streaming its stdout.
pub fn ssh_stdin_streaming(&self, command: &str, stdin: &[u8]) -> Result<()> {
let (code, _) = self.invoke(command, Some(stdin), false)?;
if code != 0 {
bail!("ssh {}: command failed (exit {code})", self.ssh_host);
}
Ok(())
}
/// Same, but return the remote exit status instead of failing: a script whose
/// exit code is part of its report needs the code, not an ssh error.
pub fn ssh_stdin_streaming_status(&self, command: &str, stdin: &[u8]) -> Result<i32> {
let (code, _) = self.invoke(command, Some(stdin), false)?;
Ok(code)
}
/// scp a local file up, creating the remote parent directory.
pub fn ship(&self, local: &Path, remote: &str) -> Result<()> {
self.ssh(&format!("mkdir -p \"$(dirname {})\"", shell_quote(remote)))?;
let status = Command::new("scp")
.arg("-F")
.arg(&self.ssh_config)
.arg("-o")
.arg("ConnectTimeout=15")
.arg(local)
.arg(format!("{}:{remote}", self.ssh_host))
.status()
.context("failed to run scp")?;
if !status.success() {
bail!("scp {} → {remote} failed", local.display());
}
Ok(())
}
/// Authorize SSH from `ip` (or the current public IP) on the security group.
///
/// Idempotent. Reports its own failures in the `die()` style and returns an
/// error so the caller can decide (bash returned 1 here).
pub fn ensure_my_ip(&self, ip: Option<&str>) -> Result<()> {
let mut aws_env: Vec<(&str, String)> = Vec::new();
if !env_present("AWS_PROFILE")
&& !env_present("AWS_ACCESS_KEY_ID")
&& aws_config_has_profile()
{
aws_env.push(("AWS_PROFILE", "akurai-ec2".to_string()));
}
let ip = match ip {
Some(value) if !value.is_empty() => value.to_string(),
_ => public_ip_lookup(),
};
if ip.is_empty() {
cry("could not determine your public IP");
bail!("could not determine your public IP");
}
if !has_command("aws") {
cry("aws CLI not available");
bail!("aws CLI not available");
}
let region = env_or("AWS_DEFAULT_REGION", || Ok("us-east-1".to_string()))?;
let described = Command::new("aws")
.envs(aws_env.iter().map(|(k, v)| (*k, v.as_str())))
.env("AWS_DEFAULT_REGION", ®ion)
.args([
"ec2",
"describe-instances",
"--filters",
&format!("Name=ip-address,Values={}", self.public_ip),
"--query",
"Reservations[].Instances[].SecurityGroups[].GroupId",
"--output",
"text",
])
.stderr(Stdio::null())
.output()
.context("failed to run aws ec2 describe-instances")?;
let sg = if described.status.success() {
String::from_utf8_lossy(&described.stdout)
.trim_end()
.to_string()
} else {
String::new()
};
if sg.is_empty() {
cry(&format!(
"could not find the security group for {} (aws creds / region?)",
self.public_ip
));
bail!("could not find the security group for {}", self.public_ip);
}
let authorized = Command::new("aws")
.envs(aws_env.iter().map(|(k, v)| (*k, v.as_str())))
.env("AWS_DEFAULT_REGION", ®ion)
.args([
"ec2",
"authorize-security-group-ingress",
"--group-id",
&sg,
"--ip-permissions",
&format!(
"IpProtocol=tcp,FromPort=22,ToPort=22,IpRanges=[{{CidrIp={ip}/32,Description=\"akurai-ec2 allow-ssh\"}}]"
),
])
.output()
.context("failed to run aws ec2 authorize-security-group-ingress")?;
let mut combined = String::from_utf8_lossy(&authorized.stdout).into_owned();
combined.push_str(&String::from_utf8_lossy(&authorized.stderr));
if authorized.status.success() {
say(&format!("SSH allowed from {ip}/32 on {sg}"));
} else if combined.contains("Duplicate") {
say(&format!("already allowed ({ip})"));
} else {
let trimmed = combined.trim_end();
cry(trimmed);
bail!("failed to authorize SSH ingress");
}
Ok(())
}
/// SSH with the auto-recovery preflight: a connection timeout is the
/// signature of a dynamic-IP drift locking us out of the security group, so
/// authorize the current IP once and retry.
fn invoke(
&self,
command: &str,
stdin: Option<&[u8]>,
capture_stdout: bool,
) -> Result<(i32, String)> {
let (code, stdout, stderr) = self.invoke_once(command, stdin, capture_stdout)?;
if code != 0
&& !SSH_RECOVERED.swap(true, Ordering::SeqCst)
&& stderr.to_lowercase().contains("timed out")
{
say("SSH timed out — auto-authorizing your current IP on the security group…");
if self.ensure_my_ip(None).is_ok() {
let (code, stdout, _) = self.invoke_once(command, stdin, capture_stdout)?;
return Ok((code, stdout));
}
}
Ok((code, stdout))
}
fn invoke_once(
&self,
command: &str,
stdin: Option<&[u8]>,
capture_stdout: bool,
) -> Result<(i32, String, String)> {
let mut cmd = Command::new("ssh");
cmd.arg("-F")
.arg(&self.ssh_config)
.arg("-o")
.arg("ConnectTimeout=15")
.arg(&self.ssh_host);
if !command.is_empty() {
cmd.arg(command);
}
cmd.stdin(if stdin.is_some() {
Stdio::piped()
} else {
Stdio::inherit()
});
cmd.stdout(if capture_stdout {
Stdio::piped()
} else {
Stdio::inherit()
});
cmd.stderr(Stdio::piped());
let mut child = cmd.spawn().context("failed to run ssh")?;
let child_stdin = child.stdin.take();
let child_stdout = child.stdout.take();
let mut child_stderr = child.stderr.take();
let mut stdout_buf: Vec<u8> = Vec::new();
let mut stderr_buf: Vec<u8> = Vec::new();
std::thread::scope(|scope| -> Result<()> {
let writer = match (child_stdin, stdin) {
(Some(mut pipe), Some(data)) => Some(scope.spawn(move || {
// A closed pipe (remote exited early) is not our error.
let _ = pipe.write_all(data);
let _ = pipe.flush();
})),
(Some(pipe), None) => {
drop(pipe);
None
}
(None, _) => None,
};
let reader = child_stdout.map(|mut pipe| {
scope.spawn(move || -> std::io::Result<Vec<u8>> {
let mut buf = Vec::new();
pipe.read_to_end(&mut buf)?;
Ok(buf)
})
});
if let Some(pipe) = child_stderr.as_mut() {
pipe.read_to_end(&mut stderr_buf)
.context("failed to read ssh stderr")?;
}
if let Some(handle) = reader {
stdout_buf = handle
.join()
.map_err(|_| anyhow!("ssh stdout reader panicked"))?
.context("failed to read ssh stdout")?;
}
if let Some(handle) = writer {
handle
.join()
.map_err(|_| anyhow!("ssh stdin writer panicked"))?;
}
Ok(())
})?;
let status = child.wait().context("failed to wait for ssh")?;
let stderr = String::from_utf8_lossy(&stderr_buf).into_owned();
// bash: `cat "$err" >&2` — the remote's diagnostics always reach the user.
if !stderr.is_empty() {
let mut handle = std::io::stderr();
let _ = handle.write_all(stderr.as_bytes());
let _ = handle.flush();
}
Ok((
status.code().unwrap_or(-1),
String::from_utf8_lossy(&stdout_buf).into_owned(),
stderr,
))
}
}
fn aws_config_has_profile() -> bool {
let Ok(home) = home() else {
return false;
};
let Ok(config) = std::fs::read_to_string(home.join(".aws").join("config")) else {
return false;
};
config.lines().any(|line| line == "[profile akurai-ec2]")
}
fn public_ip_lookup() -> String {
Command::new("curl")
.args(["-s", "--max-time", "8", "https://api.ipify.org"])
.stderr(Stdio::null())
.output()
.ok()
.filter(|out| out.status.success())
.map(|out| String::from_utf8_lossy(&out.stdout).trim().to_string())
.unwrap_or_default()
}
#[derive(Subcommand)]
pub enum Ec2Command {
/// Run a command on the box.
Ssh {
#[arg(trailing_var_arg = true, allow_hyphen_values = true)]
args: Vec<String>,
},
/// scp a file up (creates the remote parent directory).
Ship { local: PathBuf, remote: String },
/// Services, ports, nginx, disk.
Status,
/// List the running AkurAI fleet: unit, state, PID, memory, uptime, port.
#[command(alias = "ps")]
Processes { pattern: Option<String> },
/// Push a single binary and serve it via systemd on 127.0.0.1:<port>.
DeployBinary {
name: String,
bin: PathBuf,
port: u16,
appdir: Option<PathBuf>,
#[arg(long)]
json: bool,
},
/// Move encrypted Bunfork state from the source host onto the box.
MigrateBunforkState { source_ssh: Option<String> },
/// Disable and archive the source Bunfork unit after public readiness.
RetireBunforkSource {
source_ssh: Option<String>,
domain: Option<String>,
},
/// Verify a Bunfork bundle's inventory locally.
VerifyBunfork { bundle: PathBuf },
/// Manage authoritative EC2 DNS zones (mutations require --yes).
Dns {
#[arg(trailing_var_arg = true, allow_hyphen_values = true)]
args: Vec<String>,
},
/// Durable, fenced workspace lifecycle control.
Workspace {
#[arg(trailing_var_arg = true, allow_hyphen_values = true)]
args: Vec<String>,
},
/// Inventory or remove bounded EC2 app resources.
RetireApp {
#[arg(trailing_var_arg = true, allow_hyphen_values = true)]
args: Vec<String>,
},
/// Atomically deploy a verified Bunfork native bundle.
DeployBunfork {
bundle: PathBuf,
domain: Option<String>,
port: Option<u16>,
},
/// Install, run, or verify encrypted Bunfork backups.
BunforkBackup {
action: Option<String>,
retention: Option<u32>,
},
/// Install, run, or verify encrypted hourly AkurAI-Tasks backups.
TasksBackup {
action: Option<String>,
retention: Option<u32>,
},
/// Install, run, or verify encrypted Workbench snapshots.
WorkbenchBackup {
action: Option<String>,
retention: Option<u32>,
},
/// Install, run, or verify encrypted mail-server backups held on Titan.
MailBackup {
action: Option<String>,
retention: Option<u32>,
},
/// Audit mail-server security posture, backups, and pending patches.
MailAudit {
host: Option<String>,
warn_days: Option<u32>,
},
/// Report, install, or reboot into mail-server package updates.
MailUpdate { action: Option<String> },
/// List mail domains, or retire a decommissioned one.
MailDomain {
action: Option<String>,
domain: Option<String>,
},
/// Install, run, or verify HTTPS readiness and TLS-expiry monitoring.
BunforkMonitor {
action: Option<String>,
domain: Option<String>,
warn_days: Option<u32>,
recipient: Option<String>,
},
/// Reverse-proxy vhost + reload (re-asserts HTTPS if a cert exists).
NginxProxy { domain: String, port: u16 },
/// Obtain or renew a Let's Encrypt certificate.
Tls { domain: String },
/// Strict public TLS, hostname, and expiry check.
CertStatus {
domain: Option<String>,
#[arg(long, default_value_t = 30)]
warn_days: u32,
#[arg(long)]
json: bool,
},
/// Unified release engine driven by the repo's akurai-deploy.toml.
Release {
#[arg(trailing_var_arg = true, allow_hyphen_values = true)]
args: Vec<String>,
},
/// Restore <name>.prev, the previous binary.
Rollback { name: String },
/// journalctl -u <service> -n 80.
Logs { service: String },
/// Probe email client auto-setup for a domain.
MailAutoconfig {
action: Option<String>,
#[arg(trailing_var_arg = true)]
domains: Vec<String>,
},
/// Registry vs live ports: list, free [n], check <port>, drift.
Ports {
#[arg(trailing_var_arg = true, allow_hyphen_values = true)]
args: Vec<String>,
},
/// Authorize SSH from an IP (default: this host's public IP), then verify.
AllowSsh { ip: Option<String> },
/// Fan-out health check across all registered apps in AKURAI_APPS.toml.
Health {
#[arg(long)]
all: bool,
},
/// Audit every active app's production pipeline against the suite
/// deployment policy (docs/suite-deployment-policy.md), report exact
/// hosted commit and precise violations. Declaring conformance in YAML
/// is never treated as proof of a completed deployment.
FleetAudit {
#[arg(long)]
json: bool,
},
/// Cross-repo search across all AkurAI sources.
Search {
pattern: String,
#[arg(long)]
include: Option<String>,
},
/// git status across all AkurAI repos.
Repos { action: Option<String> },
/// Validate an akurai-deploy.toml schema.
ValidateConfig { path: Option<PathBuf> },
/// Read a secret from stdin, update the protected env, and restart.
SecretSync {
service: String,
key: String,
env_file: String,
},
/// Operate the dockerized MariaDB stack via on-box akurai-mariadbctl.
Mariadb {
#[arg(trailing_var_arg = true, allow_hyphen_values = true)]
args: Vec<String>,
},
/// Create an IDP client, store credentials, and merge a protected app env.
ProvisionOidcClient {
name: String,
redirect: String,
env_prefix: String,
email: String,
folder: String,
remote_env_file: Option<String>,
},
}
pub fn execute(command: Ec2Command) -> Result<()> {
let ec2 = Ec2::from_env()?;
match command {
Ec2Command::Ssh { args } => host::ssh(&ec2, &args),
Ec2Command::Ship { local, remote } => host::ship(&ec2, &local, &remote),
Ec2Command::Status => host::status(&ec2),
Ec2Command::Processes { pattern } => host::processes(&ec2, pattern.as_deref()),
Ec2Command::Logs { service } => host::logs(&ec2, &service),
Ec2Command::Ports { args } => host::ports(&ec2, &args),
Ec2Command::AllowSsh { ip } => host::allow_ssh(&ec2, ip.as_deref()),
Ec2Command::Health { all } => host::health(&ec2, all),
Ec2Command::FleetAudit { json } => host::fleet_audit(json),
Ec2Command::DeployBinary {
name,
bin,
port,
appdir,
json,
} => deploy::deploy_binary(&ec2, &name, &bin, port, appdir.as_deref(), json),
Ec2Command::Rollback { name } => deploy::rollback(&ec2, &name),
Ec2Command::NginxProxy { domain, port } => deploy::nginx_proxy(&ec2, &domain, port),
Ec2Command::Tls { domain } => deploy::tls(&ec2, &domain),
Ec2Command::CertStatus {
domain,
warn_days,
json,
} => deploy::cert_status(&ec2, domain.as_deref(), warn_days, json),
Ec2Command::Release { args } => deploy::run_release(&ec2, &args),
Ec2Command::DeployBunfork {
bundle,
domain,
port,
} => deploy::deploy_bunfork(&ec2, &bundle, domain.as_deref(), port),
Ec2Command::VerifyBunfork { bundle } => deploy::verify_bunfork(&ec2, &bundle),
Ec2Command::MigrateBunforkState { source_ssh } => {
deploy::migrate_bunfork_state(&ec2, source_ssh.as_deref())
}
Ec2Command::RetireBunforkSource { source_ssh, domain } => {
deploy::retire_bunfork_source(&ec2, source_ssh.as_deref(), domain.as_deref())
}
Ec2Command::MailAutoconfig { action, domains } => {
deploy::mail_autoconfig(&ec2, action.as_deref(), &domains)
}
Ec2Command::Dns { args } => dns::run(&ec2, &args),
Ec2Command::Workspace { args } => ops::workspace(&ec2, &args),
Ec2Command::RetireApp { args } => ops::retire_app(&ec2, &args),
Ec2Command::BunforkBackup { action, retention } => {
ops::bunfork_backup(&ec2, action.as_deref(), retention)
}
Ec2Command::TasksBackup { action, retention } => {
ops::tasks_backup(&ec2, action.as_deref(), retention)
}
Ec2Command::WorkbenchBackup { action, retention } => {
ops::workbench_backup(&ec2, action.as_deref(), retention)
}
Ec2Command::MailBackup { action, retention } => {
ops::mail_backup(&ec2, action.as_deref(), retention)
}
Ec2Command::MailAudit { host, warn_days } => {
ops::mail_audit(&ec2, host.as_deref(), warn_days)
}
Ec2Command::MailUpdate { action } => ops::mail_update(&ec2, action.as_deref()),
Ec2Command::MailDomain { action, domain } => {
ops::mail_domain(&ec2, action.as_deref(), domain.as_deref())
}
Ec2Command::BunforkMonitor {
action,
domain,
warn_days,
recipient,
} => ops::bunfork_monitor(
&ec2,
action.as_deref(),
domain.as_deref(),
warn_days,
recipient.as_deref(),
),
Ec2Command::Search { pattern, include } => ops::search(&ec2, &pattern, include.as_deref()),
Ec2Command::Repos { action } => ops::repos(&ec2, action.as_deref()),
Ec2Command::ValidateConfig { path } => ops::validate_config(&ec2, path.as_deref()),
Ec2Command::SecretSync {
service,
key,
env_file,
} => ops::secret_sync(&ec2, &service, &key, &env_file),
Ec2Command::Mariadb { args } => ops::mariadb(&ec2, &args),
Ec2Command::ProvisionOidcClient {
name,
redirect,
env_prefix,
email,
folder,
remote_env_file,
} => ops::provision_oidc_client(
&ec2,
&name,
&redirect,
&env_prefix,
&email,
&folder,
remote_env_file.as_deref(),
),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn shell_quote_neutralizes_metacharacters() {
assert_eq!(shell_quote("/opt/app"), "'/opt/app'");
assert_eq!(shell_quote("a b"), "'a b'");
// A closing quote must not escape the quoted region.
assert_eq!(shell_quote("it's; rm -rf /"), "'it'\\''s; rm -rf /'");
}
#[test]
fn pool_bounds_match_the_registry_contract() {
assert_eq!((APP_POOL_LO, APP_POOL_HI), (8094, 8199));
assert_eq!(PUBLIC_IP, "3.94.46.219");
}
}