Menu
AkurAI-Build
publicLatest change d123695f7e04f650d4b3d8c34af099a681807b3f - ec2: drop retired golfsetridak.is from mail domain defaults and repo glob by Ólafur Búi Ólafsson
//! Operational verbs ported from the `akurai-ec2` bash CLI and its Python helpers.
//!
//! Fidelity to the shell original is deliberate: argument order, defaults,
//! environment variable names, stdout text, and exit codes are reproduced
//! verbatim because seventeen live production services depend on them.
use std::{
collections::BTreeMap,
env,
fs::{self, DirBuilder, File, OpenOptions},
io::{Read, Write},
os::unix::fs::{DirBuilderExt, OpenOptionsExt, PermissionsExt},
path::{Path, PathBuf},
process::{Command, Stdio},
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use anyhow::{Context, Result, anyhow, bail};
use serde_json::{Map, Value, json};
use sha2::{Digest, Sha256};
use zeroize::Zeroizing;
use crate::ec2::{Ec2, say, shell_quote};
// ---------------------------------------------------------------------------
// shared helpers
// ---------------------------------------------------------------------------
fn home() -> Result<PathBuf> {
env::var_os("HOME")
.map(PathBuf::from)
.ok_or_else(|| anyhow!("HOME is not set"))
}
/// Resolve a helper script shipped inside this repository.
///
/// The bash CLI reached for `$HOME/Projects/AkurAI-Build/scripts/...`; a
/// deployed binary is usually installed elsewhere, so the working directory and
/// the executable's own neighbourhood are consulted first.
fn repo_script(name: &str) -> PathBuf {
let mut candidates: Vec<PathBuf> = Vec::new();
if let Ok(cwd) = env::current_dir() {
candidates.push(cwd.join("scripts").join(name));
}
if let Ok(exe) = env::current_exe()
&& let Some(directory) = exe.parent()
{
candidates.push(directory.join("scripts").join(name));
candidates.push(directory.join("..").join("scripts").join(name));
candidates.push(directory.join("../..").join("scripts").join(name));
}
if let Ok(root) = home() {
candidates.push(root.join("Projects/AkurAI-Build/scripts").join(name));
}
for candidate in &candidates {
if candidate.is_file() {
return candidate.clone();
}
}
candidates
.pop()
.unwrap_or_else(|| PathBuf::from("scripts").join(name))
}
/// The command used to re-enter this CLI for the DNS surface.
///
/// `AKURAI_EC2_CLI` keeps compatibility with the retired bash shim; otherwise
/// this binary calls itself with the `ec2` subcommand prefix.
fn self_cli() -> (PathBuf, Vec<String>) {
if let Some(cli) = env::var_os("AKURAI_EC2_CLI") {
return (PathBuf::from(cli), Vec::new());
}
let program = env::current_exe().unwrap_or_else(|_| PathBuf::from("akurai"));
(program, vec!["ec2".to_string()])
}
fn run_local(program: &Path, args: &[&str], operation: &str) -> Result<()> {
let status = Command::new(program)
.args(args)
.status()
.with_context(|| format!("failed to start {operation}"))?;
if !status.success() {
bail!("{operation} failed");
}
Ok(())
}
fn local_ok(program: &str, args: &[&str]) -> bool {
Command::new(program)
.args(args)
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.map(|status| status.success())
.unwrap_or(false)
}
fn capture_local(program: &Path, args: &[&str], operation: &str) -> Result<String> {
let output = Command::new(program)
.args(args)
.stderr(Stdio::inherit())
.output()
.with_context(|| format!("failed to start {operation}"))?;
if !output.status.success() {
bail!("{operation} failed");
}
let mut text = String::from_utf8_lossy(&output.stdout).into_owned();
while text.ends_with('\n') || text.ends_with('\r') {
text.pop();
}
Ok(text)
}
fn capture_quiet(program: &str, args: &[&str]) -> Option<String> {
let output = Command::new(program)
.args(args)
.stderr(Stdio::null())
.output()
.ok()?;
if !output.status.success() {
return None;
}
Some(String::from_utf8_lossy(&output.stdout).trim().to_string())
}
fn matches_class(value: &str, allowed: impl Fn(char) -> bool) -> bool {
!value.is_empty() && value.chars().all(allowed)
}
fn write_private(path: &Path, contents: &str) -> Result<()> {
let directory = path
.parent()
.ok_or_else(|| anyhow!("no parent directory for {}", path.display()))?;
let mut temporary = tempfile::Builder::new()
.prefix(".akurai-ec2-")
.tempfile_in(directory)
.with_context(|| format!("create a temporary file in {}", directory.display()))?;
temporary.write_all(contents.as_bytes())?;
temporary.flush()?;
fs::set_permissions(temporary.path(), fs::Permissions::from_mode(0o600))?;
temporary
.persist(path)
.map_err(|error| anyhow!("install {}: {}", path.display(), error.error))?;
Ok(())
}
fn make_private_dir(path: &Path, mode: u32) -> Result<()> {
if path.is_dir() {
return Ok(());
}
DirBuilder::new()
.recursive(true)
.mode(mode)
.create(path)
.with_context(|| format!("create {}", path.display()))
}
// ---------------------------------------------------------------------------
// time (no date dependency; the journal format is ISO-8601 UTC)
// ---------------------------------------------------------------------------
fn epoch_now() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|value| value.as_secs() as i64)
.unwrap_or(0)
}
fn nanos_now() -> u32 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|value| value.subsec_nanos())
.unwrap_or(0)
}
/// Howard Hinnant's civil-from-days.
fn civil_from_days(days: i64) -> (i64, u32, u32) {
let z = days + 719_468;
let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
let day_of_era = z - era * 146_097;
let year_of_era =
(day_of_era - day_of_era / 1460 + day_of_era / 36_524 - day_of_era / 146_096) / 365;
let year = year_of_era + era * 400;
let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100);
let mp = (5 * day_of_year + 2) / 153;
let day = (day_of_year - (153 * mp + 2) / 5 + 1) as u32;
let month = if mp < 10 { mp + 3 } else { mp - 9 } as u32;
(if month <= 2 { year + 1 } else { year }, month, day)
}
/// Howard Hinnant's days-from-civil.
fn days_from_civil(year: i64, month: u32, day: u32) -> i64 {
let year = if month <= 2 { year - 1 } else { year };
let era = if year >= 0 { year } else { year - 399 } / 400;
let year_of_era = year - era * 400;
let month = month as i64;
let day = day as i64;
let day_of_year = (153 * (if month > 2 { month - 3 } else { month + 9 }) + 2) / 5 + day - 1;
let day_of_era = year_of_era * 365 + year_of_era / 4 - year_of_era / 100 + day_of_year;
era * 146_097 + day_of_era - 719_468
}
fn iso_utc(seconds: i64) -> String {
let days = seconds.div_euclid(86_400);
let remainder = seconds.rem_euclid(86_400);
let (year, month, day) = civil_from_days(days);
format!(
"{year:04}-{month:02}-{day:02}T{:02}:{:02}:{:02}Z",
remainder / 3600,
(remainder % 3600) / 60,
remainder % 60
)
}
fn compact_stamp(seconds: i64) -> String {
let days = seconds.div_euclid(86_400);
let remainder = seconds.rem_euclid(86_400);
let (year, month, day) = civil_from_days(days);
format!(
"{year:04}{month:02}{day:02}T{:02}{:02}{:02}",
remainder / 3600,
(remainder % 3600) / 60,
remainder % 60
)
}
fn parse_iso(value: &str) -> Result<i64> {
let text = value.trim();
let text = text.strip_suffix('Z').unwrap_or(text);
let text = text.strip_suffix("+00:00").unwrap_or(text);
let (date, time) = text
.split_once('T')
.ok_or_else(|| anyhow!("invalid timestamp: {value}"))?;
let mut date_parts = date.split('-');
let year: i64 = date_parts
.next()
.and_then(|part| part.parse().ok())
.ok_or_else(|| anyhow!("invalid timestamp: {value}"))?;
let month: u32 = date_parts
.next()
.and_then(|part| part.parse().ok())
.ok_or_else(|| anyhow!("invalid timestamp: {value}"))?;
let day: u32 = date_parts
.next()
.and_then(|part| part.parse().ok())
.ok_or_else(|| anyhow!("invalid timestamp: {value}"))?;
let time = time.split('.').next().unwrap_or(time);
let mut time_parts = time.split(':');
let hour: i64 = time_parts
.next()
.and_then(|part| part.parse().ok())
.ok_or_else(|| anyhow!("invalid timestamp: {value}"))?;
let minute: i64 = time_parts
.next()
.and_then(|part| part.parse().ok())
.ok_or_else(|| anyhow!("invalid timestamp: {value}"))?;
let second: i64 = time_parts
.next()
.and_then(|part| part.parse().ok())
.unwrap_or(0);
Ok(days_from_civil(year, month, day) * 86_400 + hour * 3600 + minute * 60 + second)
}
// ---------------------------------------------------------------------------
// JSON emitted with Python's `ensure_ascii=True` so digests and files match
// ---------------------------------------------------------------------------
fn ascii_escaped(text: &str) -> String {
if text.is_ascii() {
return text.to_string();
}
let mut out = String::with_capacity(text.len());
for character in text.chars() {
if character.is_ascii() {
out.push(character);
continue;
}
let mut buffer = [0u16; 2];
for unit in character.encode_utf16(&mut buffer) {
out.push_str(&format!("\\u{unit:04x}"));
}
}
out
}
fn compact_json(value: &Value) -> String {
ascii_escaped(&value.to_string())
}
fn pretty_json(value: &Value) -> String {
ascii_escaped(&serde_json::to_string_pretty(value).unwrap_or_else(|_| "{}".to_string()))
}
// ===========================================================================
// mariadb
// ===========================================================================
const MARIADB_VERBS: [&str; 12] = [
"status",
"health",
"up",
"down",
"restart",
"mode",
"sql",
"create-db",
"create-user",
"copy-db",
"dump",
"logs",
];
/// Delegate to the on-box `akurai-mariadbctl`, or install it from the repo copy.
pub fn mariadb(ec2: &Ec2, args: &[String]) -> Result<()> {
let default = ["status".to_string()];
let args = if args.is_empty() { &default[..] } else { args };
let verb = args[0].as_str();
if MARIADB_VERBS.contains(&verb) {
let mut command = String::from("akurai-mariadbctl");
for argument in args {
command.push(' ');
command.push_str(&shell_quote(argument));
}
return ec2.ssh_streaming(&command);
}
if verb == "install" {
say("installing akurai-mariadbctl from the repo copy…");
let source = repo_script("akurai-mariadbctl");
let payload = fs::read(&source)
.with_context(|| format!("read the repo copy: {}", source.display()))?;
return ec2.ssh_stdin_streaming(
"sudo tee /usr/local/bin/akurai-mariadbctl > /dev/null && sudo chmod +x /usr/local/bin/akurai-mariadbctl && akurai-mariadbctl health",
&payload,
);
}
bail!(
"usage: akurai-ec2 mariadb <status|health|up|down|restart|mode|sql|create-db|create-user|copy-db|dump|logs|install> (see akurai-mariadbctl --help)"
)
}
// ===========================================================================
// secret-sync
// ===========================================================================
/// Read a secret from stdin, atomically update a protected env file, restart.
///
/// The secret never reaches argv, a log line, or shell history: it travels
/// stdin → remote `IFS= read -r` → `printf` into the EnvironmentFile.
pub fn secret_sync(ec2: &Ec2, service: &str, key: &str, env_file: &str) -> Result<()> {
if !matches_class(service, |c| {
c.is_ascii_alphanumeric() || matches!(c, '_' | '.' | '@' | '-')
}) {
bail!("invalid service name");
}
let mut key_chars = key.chars();
let valid_key = match key_chars.next() {
Some(first) if first.is_ascii_uppercase() => {
key_chars.all(|c| c.is_ascii_uppercase() || c.is_ascii_digit() || c == '_')
}
_ => false,
};
if !valid_key {
bail!("invalid environment key");
}
if !env_file.starts_with("/etc/") {
bail!("environment file must be below /etc");
}
let mut raw = String::new();
std::io::stdin()
.read_to_string(&mut raw)
.context("read the secret from stdin")?;
// `$(cat)` strips every trailing newline.
let secret = Zeroizing::new(raw.trim_end_matches('\n').to_string());
if secret.is_empty() {
bail!("secret-sync requires a secret on stdin");
}
if secret.contains('\n') {
bail!("secret must be one line");
}
// Values are transported verbatim (stdin -> remote `IFS= read -r` ->
// printf) into a KEY=value systemd EnvironmentFile line. Excluded:
// whitespace, quotes, backslash, $, backtick, #, ;, &, |, <, > — the write
// path would carry them, but keeping them out avoids downstream ambiguity.
if !matches_class(&secret, |c| {
c.is_ascii_alphanumeric() || "._~!@%^*+=:,/-".contains(c)
}) {
bail!("secret contains unsupported characters");
}
let remote = format!(
r#"sudo sh -c 'set -eu
tmp=$(mktemp)
trap "rm -f $tmp" EXIT
if [ -f "{env_file}" ]; then
grep -v "^[[:space:]]*\(export[[:space:]]\+\)\?{key}=" "{env_file}" >"$tmp" || true
fi
IFS= read -r value
printf "%s=%s\n" "{key}" "$value" >>"$tmp"
install -D -o root -g root -m 600 "$tmp" "{env_file}"'"#
);
let payload = Zeroizing::new(format!("{}\n", secret.as_str()).into_bytes());
if ec2.ssh_stdin_streaming(&remote, &payload).is_err() {
bail!("failed to install secret");
}
drop(payload);
drop(secret);
if ec2
.ssh_streaming(&format!(
"sudo systemctl restart '{service}' && sudo systemctl is-active --quiet '{service}'"
))
.is_err()
{
bail!("secret installed but {service} failed to restart");
}
say(&format!("secret synchronized and {service} active"));
Ok(())
}
// ===========================================================================
// validate-config
// ===========================================================================
/// Minimal top-level TOML reader: enough for the flat `akurai-deploy.toml`
/// schema this verb validates, with no new dependency.
fn top_level_keys(text: &str) -> BTreeMap<String, (String, bool)> {
let mut keys = BTreeMap::new();
for line in text.lines() {
let trimmed = line.trim();
if trimmed.is_empty() || trimmed.starts_with('#') {
continue;
}
if trimmed.starts_with('[') {
break; // only the top-level table matters here
}
let Some((name, raw)) = trimmed.split_once('=') else {
continue;
};
let name = name.trim().trim_matches('"').to_string();
let mut value = String::new();
let mut quoted = false;
let mut in_string = false;
for character in raw.trim().chars() {
match character {
'"' | '\'' => {
quoted = true;
in_string = !in_string;
}
'#' if !in_string => break,
_ => value.push(character),
}
}
keys.insert(name, (value.trim().to_string(), quoted));
}
keys
}
/// Validate an `akurai-deploy.toml`. The bash arm swallowed the validator's
/// exit code (`2>&1 || true`), so a failing document still exits 0.
pub fn validate_config(_ec2: &Ec2, path: Option<&Path>) -> Result<()> {
let path = path.map(Path::to_path_buf).unwrap_or_else(|| {
env::current_dir()
.unwrap_or_else(|_| PathBuf::from("."))
.join("akurai-deploy.toml")
});
if !path.is_file() {
bail!("not found: {}", path.display());
}
let text = match fs::read_to_string(&path) {
Ok(text) => text,
Err(error) => {
println!("{error}");
return Ok(());
}
};
let config = top_level_keys(&text);
let mut errors: Vec<String> = Vec::new();
for required in ["name", "port"] {
if !config.contains_key(required) {
errors.push(format!("missing required key: {required}"));
}
}
if let Some((raw, quoted)) = config.get("port") {
let numeric = if *quoted {
None
} else {
raw.parse::<i64>().ok()
};
match numeric {
Some(port) if (1024..=65535).contains(&port) => {}
_ => errors.push(format!("invalid port: {raw} (must be 1024-65535)")),
}
}
if let Some((domain, _)) = config.get("domain")
&& domain != "-"
&& !domain.contains('.')
{
errors.push(format!("domain \"{domain}\" looks invalid"));
}
if !errors.is_empty() {
println!("VALIDATION FAILED:");
for error in errors {
println!(" - {error}");
}
return Ok(());
}
let name = config.get("name").map(|(v, _)| v.as_str()).unwrap_or("?");
let domain = config.get("domain").map(|(v, _)| v.as_str()).unwrap_or("-");
let port = config.get("port").map(|(v, _)| v.as_str()).unwrap_or("?");
let build = config
.get("build")
.map(|(v, _)| v.as_str())
.unwrap_or("(shared binary)");
let build: String = build.chars().take(60).collect();
println!("✅ {name} — config valid");
println!(" domain: {domain}");
println!(" port: {port}");
println!(" build: {build}");
Ok(())
}
// ===========================================================================
// repos / search
// ===========================================================================
fn project_dirs(patterns: &[&str]) -> Result<Vec<PathBuf>> {
let projects = home()?.join("Projects");
let mut found = Vec::new();
for pattern in patterns {
let full = projects.join(pattern);
let Some(text) = full.to_str() else { continue };
let Ok(entries) = glob::glob(text) else {
continue;
};
for entry in entries.flatten() {
found.push(entry);
}
}
Ok(found)
}
/// `repos status` — branch, dirt, and upstream drift across the AkurAI repos.
pub fn repos(_ec2: &Ec2, action: Option<&str>) -> Result<()> {
match action.unwrap_or("status") {
"status" => {}
_ => bail!("usage: akurai-ec2 repos status"),
}
println!("=== AkurAI Repos Status ===");
for directory in project_dirs(&["AkurAI-*", "akurai-*"])? {
if !directory.join(".git").exists() {
continue;
}
let Some(path) = directory.to_str() else {
continue;
};
let name = directory
.file_name()
.and_then(|value| value.to_str())
.unwrap_or(path)
.to_string();
let branch = capture_quiet("git", &["-C", path, "rev-parse", "--abbrev-ref", "HEAD"])
.unwrap_or_else(|| "?".to_string());
let dirty = capture_quiet("git", &["-C", path, "status", "--porcelain"])
.map(|text| {
if text.is_empty() {
0
} else {
text.lines().count()
}
})
.unwrap_or(0);
let ahead = capture_quiet(
"git",
&["-C", path, "rev-list", "--count", "@{upstream}..HEAD"],
)
.unwrap_or_else(|| "0".to_string());
let behind = capture_quiet(
"git",
&["-C", path, "rev-list", "--count", "HEAD..@{upstream}"],
)
.unwrap_or_else(|| "0".to_string());
println!(" {name:<35} {branch:<22} dirty={dirty:<3} ahead={ahead:<3} behind={behind:<3}");
}
Ok(())
}
/// Cross-repo source search. The bash original passed grep's `--include` to
/// ripgrep, which rejects that flag, so every search silently found nothing;
/// the working equivalent (`--glob`) is used here.
pub fn search(_ec2: &Ec2, pattern: &str, include: Option<&str>) -> Result<()> {
let include = include.unwrap_or("*.rs");
println!("=== Searching AkurAI repos for: {pattern} ===");
for directory in project_dirs(&["AkurAI-*", "akurai-*"])? {
if !directory.is_dir() {
continue;
}
let Some(path) = directory.to_str() else {
continue;
};
let name = directory
.file_name()
.and_then(|value| value.to_str())
.unwrap_or(path)
.to_string();
let source = format!("{path}/src");
let crates = format!("{path}/crates");
let Some(matches) =
capture_quiet("rg", &["-l", "--glob", include, pattern, &source, &crates])
else {
continue;
};
let files: Vec<&str> = matches
.lines()
.filter(|line| !line.is_empty())
.take(5)
.collect();
if files.is_empty() {
continue;
}
println!(" [{name}]");
let prefix = format!("{path}/");
for file in files {
println!(" {}", file.strip_prefix(&prefix).unwrap_or(file));
}
}
Ok(())
}
// ===========================================================================
// provision-oidc-client
// ===========================================================================
/// Create an IDP client, store credentials in PassVault, merge a protected env.
pub fn provision_oidc_client(
_ec2: &Ec2,
name: &str,
redirect: &str,
env_prefix: &str,
email: &str,
folder: &str,
remote_env_file: Option<&str>,
) -> Result<()> {
let script = repo_script("provision-app-oidc.py");
if !script.is_file() {
bail!("provision helper not found: {}", script.display());
}
let mut command = Command::new(&script);
command.args([name, redirect, env_prefix, email, folder]);
if let Some(remote) = remote_env_file {
command.arg(remote);
}
let status = command
.status()
.with_context(|| format!("failed to start {}", script.display()))?;
if !status.success() {
std::process::exit(status.code().unwrap_or(1));
}
Ok(())
}
// ===========================================================================
// retire-app
// ===========================================================================
const RETIRE_APP_REMOTE: &str = r#"set -euo pipefail
name="$1"; domain="$2"; action="$3"; delete_data="$4"; delete_user="$5"; delete_cert="$6"
exists() { sudo test -e "$1" && printf true || printf false; }
unit=$(systemctl list-unit-files "$name.service" --no-legend 2>/dev/null | grep -q "$name.service" && printf true || printf false)
active=$(systemctl is-active --quiet "$name.service" 2>/dev/null && printf true || printf false)
user=$(id "$name" >/dev/null 2>&1 && printf true || printf false)
printf '{"name":"%s","domain":"%s","unit":%s,"active":%s,"app":%s,"data":%s,"config":%s,"backup":%s,"nginx_available":%s,"nginx_enabled":%s,"user":%s}\n' \
"$name" "$domain" "$unit" "$active" "$(exists "/opt/$name")" "$(exists "/var/lib/$name")" \
"$(exists "/etc/$name")" "$(exists "/var/backups/$name")" \
"$(exists "/etc/nginx/sites-available/$domain")" "$(exists "/etc/nginx/sites-enabled/$domain")" "$user"
[ "$action" = inventory ] && exit 0
sudo systemctl disable --now "$name.service" 2>/dev/null || true
sudo rm -f "/etc/systemd/system/$name.service"
sudo rm -rf "/opt/$name"
sudo rm -f "/etc/nginx/sites-enabled/$domain" "/etc/nginx/sites-available/$domain"
if [ "$delete_data" = true ]; then sudo rm -rf "/var/lib/$name" "/etc/$name" "/var/backups/$name"; fi
if [ "$delete_cert" = true ]; then sudo certbot delete --cert-name "$domain" --non-interactive 2>/dev/null || true; fi
if [ "$delete_user" = true ]; then sudo userdel "$name" 2>/dev/null || true; sudo groupdel "$name" 2>/dev/null || true; fi
sudo systemctl daemon-reload
sudo systemctl reset-failed
sudo nginx -t >/dev/null
sudo systemctl reload nginx
! systemctl list-unit-files "$name.service" --no-legend 2>/dev/null | grep -q "$name.service"
! sudo test -e "/opt/$name"
! sudo test -e "/etc/nginx/sites-enabled/$domain"
! sudo test -e "/etc/nginx/sites-available/$domain"
[ "$delete_data" != true ] || { ! sudo test -e "/var/lib/$name" && ! sudo test -e "/etc/$name" && ! sudo test -e "/var/backups/$name"; }
[ "$delete_user" != true ] || ! id "$name" >/dev/null 2>&1
printf '{"retired":"%s","domain":"%s","ok":true}\n' "$name" "$domain"
"#;
fn valid_app_name(value: &str) -> bool {
let mut chars = value.chars();
let Some(first) = chars.next() else {
return false;
};
if !(first.is_ascii_lowercase() || first.is_ascii_digit()) {
return false;
}
value.len() <= 63 && chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
}
fn valid_retire_domain(value: &str) -> bool {
let length = value.chars().count();
if !(3..=254).contains(&length) {
return false;
}
let bytes: Vec<char> = value.chars().collect();
let first = bytes[0];
let last = bytes[length - 1];
if !(first.is_ascii_alphanumeric() && last.is_ascii_alphanumeric()) {
return false;
}
bytes[1..length - 1]
.iter()
.all(|c| c.is_ascii_alphanumeric() || *c == '.' || *c == '-')
}
/// Inventory or retire one EC2 application with bounded, explicit deletion.
pub fn retire_app(ec2: &Ec2, args: &[String]) -> Result<()> {
let mut name: Option<String> = None;
let mut domain: Option<String> = None;
let mut inventory = false;
let mut apply = false;
let mut delete_data = false;
let mut delete_user = false;
let mut delete_certificate = false;
let mut delete_dns = false;
let mut dns_type = "A".to_string();
let mut yes = false;
let mut index = 0;
while index < args.len() {
let argument = args[index].as_str();
let mut value_for = |flag: &str| -> Result<String> {
index += 1;
args.get(index)
.cloned()
.ok_or_else(|| anyhow!("argument {flag}: expected one argument"))
};
match argument {
"--domain" => domain = Some(value_for("--domain")?),
"--dns-type" => dns_type = value_for("--dns-type")?,
"--inventory" => inventory = true,
"--apply" => apply = true,
"--delete-data" => delete_data = true,
"--delete-user" => delete_user = true,
"--delete-certificate" => delete_certificate = true,
"--delete-dns" => delete_dns = true,
"--yes" => yes = true,
other if other.starts_with('-') => {
eprintln!(
"usage: akurai-ec2 retire-app NAME --domain DOMAIN [--inventory | --apply]"
);
eprintln!("akurai-ec2 retire-app: error: unrecognized arguments: {other}");
std::process::exit(2);
}
other => {
if name.is_some() {
eprintln!(
"usage: akurai-ec2 retire-app NAME --domain DOMAIN [--inventory | --apply]"
);
eprintln!("akurai-ec2 retire-app: error: unrecognized arguments: {other}");
std::process::exit(2);
}
name = Some(other.to_string());
}
}
index += 1;
}
if inventory && apply {
eprintln!(
"akurai-ec2 retire-app: error: argument --apply: not allowed with argument --inventory"
);
std::process::exit(2);
}
let Some(name) = name else {
eprintln!("akurai-ec2 retire-app: error: the following arguments are required: name");
std::process::exit(2);
};
let Some(domain) = domain else {
eprintln!("akurai-ec2 retire-app: error: the following arguments are required: --domain");
std::process::exit(2);
};
if !valid_app_name(&name) {
eprintln!("invalid application name: '{name}'");
std::process::exit(1);
}
let domain = domain.trim_end_matches('.').to_string();
if !valid_retire_domain(&domain) {
eprintln!("invalid domain: '{domain}'");
std::process::exit(1);
}
if apply && !yes {
eprintln!("retire-app --apply requires --yes");
std::process::exit(1);
}
if !apply && (delete_data || delete_user || delete_certificate || delete_dns) {
eprintln!("deletion flags require --apply");
std::process::exit(1);
}
let action = if apply { "apply" } else { "inventory" };
let remote = format!(
"bash -s -- {name} {domain} {action} {delete_data} {delete_user} {delete_certificate}"
);
if ec2
.ssh_stdin_streaming(&remote, RETIRE_APP_REMOTE.as_bytes())
.is_err()
{
std::process::exit(1);
}
if apply && delete_dns {
let record_type = dns_type.to_uppercase();
let (program, prefix) = self_cli();
let mut delete = Command::new(&program);
delete.args(&prefix);
delete.args(["dns", "delete", &domain, "--type", &record_type, "--yes"]);
let status = delete
.status()
.with_context(|| format!("failed to start {}", program.display()))?;
if !status.success() {
std::process::exit(status.code().unwrap_or(1));
}
let mut verify = Command::new(&program);
verify.args(&prefix);
verify.args(["dns", "verify", &domain, "--type", &record_type, "--absent"]);
let status = verify
.status()
.with_context(|| format!("failed to start {}", program.display()))?;
if !status.success() {
eprintln!(
"{{\"ok\": false, \"error\": \"public DNS still resolves; propagation may be pending\"}}"
);
std::process::exit(status.code().unwrap_or(1));
}
}
Ok(())
}
// ===========================================================================
// workspace — durable, fenced lifecycle boundary
// ===========================================================================
const WORKSPACE_ACTIONS: [&str; 5] = [
"provision",
"deprovision",
"upgrade",
"health-register",
"dns-handoff",
];
const LEASE_SECONDS: i64 = 900;
struct OperationStore {
directory: PathBuf,
path: PathBuf,
lock_path: PathBuf,
}
impl OperationStore {
fn new() -> Result<Self> {
let directory = match env::var_os("AKURAI_WORKSPACE_STATE_DIR") {
Some(value) if !value.is_empty() => {
let raw = PathBuf::from(value);
if let Ok(stripped) = raw.strip_prefix("~") {
home()?.join(stripped)
} else {
raw
}
}
_ => {
let base = match env::var_os("XDG_STATE_HOME") {
Some(value) if !value.is_empty() => PathBuf::from(value),
_ => home()?.join(".local/state"),
};
base.join("akurai-ec2")
}
};
Ok(Self {
path: directory.join("workspace-lifecycle.json"),
lock_path: directory.join("workspace-lifecycle.lock"),
directory,
})
}
fn load(&self) -> Result<Value> {
if !self.path.exists() {
return Ok(json!({"schema": 1, "operations": {}}));
}
let text = fs::read_to_string(&self.path)
.map_err(|error| anyhow!("workspace lifecycle journal is unreadable: {error}"))?;
let value: Value = serde_json::from_str(&text)
.map_err(|error| anyhow!("workspace lifecycle journal is unreadable: {error}"))?;
let usable = value.get("schema") == Some(&json!(1))
&& value.get("operations").map(Value::is_object) == Some(true);
if !value.is_object() || !usable {
bail!("workspace lifecycle journal has an unsupported schema");
}
Ok(value)
}
fn save(&self, state: &Value) -> Result<()> {
make_private_dir(&self.directory, 0o700)?;
write_private(&self.path, &format!("{}\n", pretty_json(state)))
}
fn lock(&self, exclusive: bool) -> Result<File> {
make_private_dir(&self.directory, 0o700)?;
let file = OpenOptions::new()
.append(true)
.read(true)
.create(true)
.mode(0o600)
.open(&self.lock_path)
.with_context(|| format!("open {}", self.lock_path.display()))?;
fs::set_permissions(&self.lock_path, fs::Permissions::from_mode(0o600))?;
if exclusive {
file.lock()?;
} else {
file.lock_shared()?;
}
Ok(file)
}
fn mutate<T>(&self, operation: impl FnOnce(&mut Value) -> Result<T>) -> Result<T> {
let _guard = self.lock(true)?;
let mut state = self.load()?;
let result = operation(&mut state)?;
self.save(&state)?;
Ok(result)
}
fn read<T>(&self, operation: impl FnOnce(&Value) -> Result<T>) -> Result<T> {
let _guard = self.lock(false)?;
let state = self.load()?;
operation(&state)
}
}
fn now_iso() -> String {
iso_utc(epoch_now())
}
fn lease_expiry() -> String {
iso_utc(epoch_now() + LEASE_SECONDS)
}
fn append_audit(record: &mut Value, event: &str, details: Vec<(&str, Value)>) {
let entry_index = record
.get("audit")
.and_then(Value::as_array)
.map(Vec::len)
.unwrap_or(0);
let mut entry = Map::new();
entry.insert("sequence".into(), json!(entry_index + 1));
entry.insert("at".into(), json!(now_iso()));
entry.insert("event".into(), json!(event));
for (key, value) in details {
entry.insert(key.to_string(), value);
}
match record.get_mut("audit").and_then(Value::as_array_mut) {
Some(audit) => audit.push(Value::Object(entry)),
None => {
if let Some(object) = record.as_object_mut() {
object.insert("audit".into(), json!([Value::Object(entry)]));
}
}
}
}
fn operation_id(action: &str, idempotency_key: &str, request: &Value) -> String {
let source = compact_json(&json!({
"action": action,
"idempotency_key": idempotency_key,
"request": request,
}));
let digest = Sha256::digest(source.as_bytes());
let hex = digest
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>();
format!("wop_{}", &hex[..24])
}
fn validate_workspace_name(label: &str, value: &str) -> Result<String> {
let trimmed = value.trim();
let printable = !trimmed.is_empty()
&& trimmed.chars().count() <= 200
&& !trimmed
.chars()
.any(|c| (c as u32) < 0x20 || (c as u32) == 0x7f);
if !printable {
bail!("{label} must contain 1 through 200 printable characters");
}
Ok(trimmed.to_string())
}
fn validate_workspace_domain(value: &str) -> Result<String> {
let domain = value.trim_end_matches('.').to_lowercase();
let characters: Vec<char> = domain.chars().collect();
let shaped = match characters.len() {
0 => false,
1 => characters[0].is_ascii_alphanumeric(),
length => {
length <= 253
&& characters[0].is_ascii_alphanumeric()
&& characters[length - 1].is_ascii_alphanumeric()
&& characters[1..length - 1]
.iter()
.all(|c| c.is_ascii_alphanumeric() || *c == '.' || *c == '-')
}
};
if !shaped || domain.contains("..") || !domain.contains('.') {
bail!("domain must be a fully qualified DNS name");
}
Ok(domain)
}
fn pending_steps(action: &str) -> Value {
match action {
"provision" => json!([
{"name": "drive_workspace", "state": "pending", "reason": "Drive lacks an authenticated idempotent workspace create/read/delete write contract."},
{"name": "notes_workspace", "state": "pending", "reason": "Notes lacks an authenticated workspace tenancy write contract."},
{"name": "tasks_workspace", "state": "pending", "reason": "Tasks lacks a source-owned workspace tenancy write contract."},
{"name": "idp_tenant", "state": "pending", "reason": "IDP lacks an authenticated idempotent tenant create/read/delete write contract; provision-oidc-client only creates app clients."},
{"name": "workspace_install", "state": "pending", "reason": "The EC2 deploy surface is app-scoped, not workspace-scoped; no installation target contract exists."}
]),
"deprovision" => json!([
{"name": "retention_transition", "state": "pending", "reason": "No source-owned cancel-at-period-end or retention write contract exists."},
{"name": "source_resource_delete", "state": "pending", "reason": "No source-owned Drive, Notes, Tasks, or IDP deletion write contract exists."}
]),
"upgrade" => json!([
{"name": "workspace_upgrade", "state": "pending", "reason": "The maintained EC2 deployment contract has no workspace resource selector or installed-version readback."}
]),
"health-register" => json!([
{"name": "monitor_registration", "state": "pending", "reason": "The maintained Monitor MCP exposes reads only; it has no workspace health-registration write contract."}
]),
_ => json!([
{"name": "dns_handoff", "state": "pending", "reason": "Requires explicit --apply."}
]),
}
}
fn next_action(record: &Value) -> &'static str {
let state = record.get("state").and_then(Value::as_str).unwrap_or("");
let action = record.get("action").and_then(Value::as_str).unwrap_or("");
match state {
"approval_required" => {
"Approve the irreversible deletion in a separate command with workspace approve-delete."
}
"completed" => "No action required; the source-owned DNS handoff is complete.",
"compensation_pending" => {
"Inspect the recorded prior DNS resource, then resume after the source is reachable."
}
_ if action == "dns-handoff" => {
"Run workspace resume <operation_id> --apply only after confirming the DNS target."
}
_ => {
"Implement the listed source-owned write contracts, then resume with the returned operation_id and fence."
}
}
}
fn response(record: &Value) -> Value {
let mut result = record.clone();
let action = next_action(&result);
if let Some(object) = result.as_object_mut() {
object.insert("next_action".into(), json!(action));
}
result
}
fn find_operation<'a>(state: &'a Value, op_id: &str) -> Result<&'a Value> {
state
.get("operations")
.and_then(|operations| operations.get(op_id))
.ok_or_else(|| anyhow!("workspace operation not found: {op_id}"))
}
fn find_operation_mut<'a>(state: &'a mut Value, op_id: &str) -> Result<&'a mut Value> {
state
.get_mut("operations")
.and_then(|operations| operations.get_mut(op_id))
.ok_or_else(|| anyhow!("workspace operation not found: {op_id}"))
}
#[derive(Default)]
struct RequestArgs {
action: String,
idempotency_key: String,
organization: String,
workspace: String,
domain: Option<String>,
record_type: Option<String>,
record_value: Option<String>,
zone: Option<String>,
provider: String,
ttl: i64,
irreversible: bool,
apply: bool,
}
fn request_payload(args: &RequestArgs) -> Result<Value> {
let mut payload = Map::new();
payload.insert(
"organization".into(),
json!(validate_workspace_name("organization", &args.organization)?),
);
payload.insert(
"workspace".into(),
json!(validate_workspace_name("workspace", &args.workspace)?),
);
if args.action == "dns-handoff" {
let (Some(domain), Some(record_type), Some(record_value)) = (
args.domain.as_ref(),
args.record_type.as_ref(),
args.record_value.as_ref(),
) else {
bail!("dns-handoff requires --domain, --record-type, and --record-value");
};
let record_type = record_type.to_uppercase();
if !["A", "AAAA", "CNAME", "TXT"].contains(&record_type.as_str()) {
bail!("record type must be A, AAAA, CNAME, or TXT");
}
if record_value.contains('\n') || record_value.trim().is_empty() {
bail!("record value must be one non-empty line");
}
if !(60..=86_400).contains(&args.ttl) {
bail!("DNS TTL must be from 60 through 86400 seconds");
}
payload.insert("domain".into(), json!(validate_workspace_domain(domain)?));
payload.insert("record_type".into(), json!(record_type));
payload.insert("record_value".into(), json!(record_value));
payload.insert("ttl".into(), json!(args.ttl));
payload.insert(
"zone".into(),
match args.zone.as_ref() {
Some(zone) => json!(validate_workspace_domain(zone)?),
None => Value::Null,
},
);
payload.insert("provider".into(), json!(args.provider));
}
if args.action == "deprovision" {
payload.insert("irreversible".into(), json!(args.irreversible));
}
Ok(Value::Object(payload))
}
fn valid_idempotency_key(value: &str) -> bool {
let mut characters = value.chars();
let Some(first) = characters.next() else {
return false;
};
first.is_ascii_alphanumeric()
&& value.chars().count() <= 200
&& characters.all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | ':' | '-'))
}
fn create_or_get(store: &OperationStore, args: &RequestArgs) -> Result<Value> {
if !valid_idempotency_key(&args.idempotency_key) {
bail!("idempotency key must be 1 through 200 URL-safe characters");
}
if args.apply && args.action != "dns-handoff" {
bail!("--apply is currently supported only for dns-handoff");
}
let payload = request_payload(args)?;
let op_id = operation_id(&args.action, &args.idempotency_key, &payload);
let record = store.mutate(|state| {
let operations = state
.get_mut("operations")
.and_then(Value::as_object_mut)
.ok_or_else(|| anyhow!("workspace lifecycle journal has an unsupported schema"))?;
let existing_conflict = operations
.iter()
.find(|(_, existing)| {
existing.get("idempotency_key").and_then(Value::as_str)
== Some(args.idempotency_key.as_str())
})
.map(|(id, _)| id.clone());
if let Some(existing_id) = existing_conflict {
if existing_id == op_id {
let fence = operations
.get(&existing_id)
.and_then(|record| record.get("fence"))
.cloned()
.unwrap_or(json!(0));
let existing = operations
.get_mut(&existing_id)
.ok_or_else(|| anyhow!("workspace operation not found: {existing_id}"))?;
append_audit(existing, "idempotency_replayed", vec![("fence", fence)]);
return Ok(existing.clone());
}
bail!(
"idempotency key already owns {existing_id}; use a distinct key for a different request"
);
}
let irreversible = payload
.get("irreversible")
.and_then(Value::as_bool)
.unwrap_or(false);
let state_value = if args.action == "deprovision" && irreversible {
"approval_required"
} else {
"pending"
};
let mut record = json!({
"operation_id": op_id,
"action": args.action,
"idempotency_key": args.idempotency_key,
"request": payload,
"state": state_value,
"fence": 0,
"lease_expires_at": Value::Null,
"resource_ids": {},
"steps": pending_steps(&args.action),
"audit": [],
});
if args.action == "deprovision"
&& let Some(object) = record.as_object_mut()
{
object.insert(
"irreversible_delete".into(),
json!({
"requested": irreversible,
"approval_state": if irreversible { "required" } else { "not_requested" },
"approval_token_sha256": Value::Null,
}),
);
}
append_audit(&mut record, "requested", vec![("state", json!(state_value))]);
operations.insert(op_id.clone(), record.clone());
Ok(record)
})?;
let record = if args.apply
&& record.get("action").and_then(Value::as_str) == Some("dns-handoff")
&& record.get("state").and_then(Value::as_str) != Some("completed")
{
execute_dns(store, &op_id)?
} else {
record
};
Ok(response(&record))
}
fn approve_delete(store: &OperationStore, op_id: &str, approval_token: &str) -> Result<Value> {
if approval_token.chars().count() < 16 {
bail!("approval token must contain at least 16 characters");
}
let digest = Sha256::digest(approval_token.as_bytes())
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>();
let record = store.mutate(|state| {
let record = find_operation_mut(state, op_id)?;
let irreversible = record
.get("request")
.and_then(|request| request.get("irreversible"))
.and_then(Value::as_bool)
.unwrap_or(false);
if record.get("action").and_then(Value::as_str) != Some("deprovision") || !irreversible {
bail!("only an irreversible deprovision operation can be approved");
}
let approved = record
.get("irreversible_delete")
.and_then(|deletion| deletion.get("approval_state"))
.and_then(Value::as_str)
== Some("approved");
if approved {
let recorded = record
.get("irreversible_delete")
.and_then(|deletion| deletion.get("approval_token_sha256"))
.and_then(Value::as_str)
.unwrap_or("");
if recorded != digest {
bail!("a different irreversible-delete approval was already recorded");
}
append_audit(record, "approval_idempotency_replayed", Vec::new());
return Ok(record.clone());
}
if let Some(deletion) = record
.get_mut("irreversible_delete")
.and_then(Value::as_object_mut)
{
deletion.insert("approval_state".into(), json!("approved"));
deletion.insert("approval_token_sha256".into(), json!(digest));
}
if let Some(object) = record.as_object_mut() {
object.insert("state".into(), json!("pending"));
}
append_audit(record, "irreversible_delete_approved", Vec::new());
Ok(record.clone())
})?;
Ok(response(&record))
}
fn resume(store: &OperationStore, op_id: &str, apply: bool) -> Result<Value> {
let record = store.mutate(|state| {
let record = find_operation_mut(state, op_id)?;
let current = record
.get("state")
.and_then(Value::as_str)
.unwrap_or("")
.to_string();
if current == "completed" {
let fence = record.get("fence").cloned().unwrap_or(json!(0));
append_audit(
record,
"resume_idempotency_replayed",
vec![("fence", fence)],
);
return Ok(record.clone());
}
if current == "approval_required" {
bail!("irreversible deletion requires workspace approve-delete before resume");
}
if current == "running"
&& let Some(expiry) = record.get("lease_expires_at").and_then(Value::as_str)
&& parse_iso(expiry)? > epoch_now()
{
bail!("operation has a live execution lease; wait for it or retry after it expires");
}
let fence = record.get("fence").and_then(Value::as_i64).unwrap_or(0) + 1;
if let Some(object) = record.as_object_mut() {
object.insert("fence".into(), json!(fence));
object.insert("lease_expires_at".into(), Value::Null);
object.insert("state".into(), json!("pending"));
}
append_audit(record, "resumed", vec![("fence", json!(fence))]);
Ok(record.clone())
})?;
let record = if apply {
if record.get("action").and_then(Value::as_str) != Some("dns-handoff") {
bail!("--apply is currently supported only for dns-handoff");
}
execute_dns(store, op_id)?
} else {
record
};
Ok(response(&record))
}
fn dns_cli_args(request: &Value, action: &str) -> Vec<String> {
let string = |key: &str| -> String {
request
.get(key)
.and_then(Value::as_str)
.unwrap_or_default()
.to_string()
};
let mut args = vec!["dns".to_string(), action.to_string(), string("domain")];
if action == "upsert" {
let ttl = request
.get("ttl")
.and_then(Value::as_i64)
.unwrap_or(900)
.to_string();
args.extend([
string("record_type"),
string("record_value"),
"--ttl".to_string(),
ttl,
]);
} else {
args.extend(["--type".to_string(), string("record_type")]);
if action == "delete" {
args.extend(["--value".to_string(), string("record_value")]);
}
}
if let Some(zone) = request.get("zone").and_then(Value::as_str)
&& !zone.is_empty()
{
args.extend(["--zone".to_string(), zone.to_string()]);
}
args.extend(["--provider".to_string(), string("provider")]);
if action == "upsert" || action == "delete" {
args.push("--yes".to_string());
}
args
}
fn run_dns_cli(args: &[String]) -> Result<Value> {
let (program, prefix) = self_cli();
let label = args
.iter()
.take(2)
.cloned()
.collect::<Vec<String>>()
.join(" ");
let output = Command::new(&program)
.args(&prefix)
.args(args)
.output()
.map_err(|error| anyhow!("{label}: unable to execute {}: {error}", program.display()))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string();
let message = if !stderr.is_empty() {
stderr
} else if !stdout.is_empty() {
stdout
} else {
"akurai-ec2 command failed".to_string()
};
bail!("{label}: {message}");
}
serde_json::from_slice(&output.stdout).map_err(|_| anyhow!("{label}: expected JSON output"))
}
fn matching_record(
payload: &Value,
name: &str,
record_type: &str,
record_value: Option<&str>,
) -> Result<Option<Value>> {
let records = payload.get("records").cloned().unwrap_or(json!([]));
let Some(records) = records.as_array() else {
bail!("DNS source returned an invalid records payload");
};
let wanted = name.trim_end_matches('.').to_lowercase();
let candidates: Vec<Value> = records
.iter()
.filter(|value| value.is_object())
.filter(|value| {
value
.get("fqdn")
.and_then(Value::as_str)
.unwrap_or("")
.trim_end_matches('.')
.to_lowercase()
== wanted
&& value.get("type").and_then(Value::as_str) == Some(record_type)
&& match record_value {
None => true,
Some(expected) => value.get("value").and_then(Value::as_str) == Some(expected),
}
})
.cloned()
.collect();
if candidates.len() > 1 {
bail!(
"DNS handoff refuses multiple same-type records because exact compensation is ambiguous"
);
}
Ok(candidates.into_iter().next())
}
fn dns_lookup(request: &Value) -> Result<(Value, Option<Value>)> {
let payload = run_dns_cli(&dns_cli_args(request, "get"))?;
let name = request
.get("domain")
.and_then(Value::as_str)
.unwrap_or_default();
let record_type = request
.get("record_type")
.and_then(Value::as_str)
.unwrap_or_default();
let found = matching_record(&payload, name, record_type, None)?;
Ok((payload, found))
}
fn set_running(store: &OperationStore, op_id: &str) -> Result<(Value, i64)> {
store.mutate(|state| {
let record = find_operation_mut(state, op_id)?;
let current = record
.get("state")
.and_then(Value::as_str)
.unwrap_or("")
.to_string();
if current == "completed" {
let fence = record.get("fence").and_then(Value::as_i64).unwrap_or(0);
return Ok((record.clone(), fence));
}
if current == "approval_required" {
bail!("irreversible deletion requires explicit approval");
}
if current == "running"
&& let Some(expiry) = record.get("lease_expires_at").and_then(Value::as_str)
&& parse_iso(expiry)? > epoch_now()
{
bail!("operation already has a live execution lease");
}
let fence = record.get("fence").and_then(Value::as_i64).unwrap_or(0) + 1;
if let Some(object) = record.as_object_mut() {
object.insert("fence".into(), json!(fence));
object.insert("state".into(), json!("running"));
object.insert("lease_expires_at".into(), json!(lease_expiry()));
}
if let Some(step) = record
.get_mut("steps")
.and_then(Value::as_array_mut)
.and_then(|steps| steps.first_mut())
.and_then(Value::as_object_mut)
{
step.insert("state".into(), json!("running"));
}
append_audit(record, "dns_handoff_started", vec![("fence", json!(fence))]);
Ok((record.clone(), fence))
})
}
fn update_for_fence(
store: &OperationStore,
op_id: &str,
fence: i64,
update: impl FnOnce(&mut Value) -> Result<()>,
) -> Result<Value> {
store.mutate(|state| {
let record = find_operation_mut(state, op_id)?;
if record.get("fence").and_then(Value::as_i64) != Some(fence) {
bail!("stale execution fence; its source result was not recorded");
}
update(record)?;
Ok(record.clone())
})
}
fn compensate_dns(request: &Value, prior: Option<&Value>) -> (bool, String) {
let attempt = || -> Result<String> {
let (_, current) = dns_lookup(request)?;
if let Some(prior) = prior {
if let Some(current) = current.as_ref()
&& current.get("value") == prior.get("value")
{
return Ok("original DNS record remained in place".to_string());
}
let mut restore = request.clone();
if let Some(object) = restore.as_object_mut() {
let value = match prior.get("value") {
Some(Value::String(text)) => text.clone(),
Some(other) => other.to_string(),
None => String::new(),
};
object.insert("record_value".into(), json!(value));
let ttl = prior
.get("ttl")
.and_then(|value| {
value
.as_i64()
.or_else(|| value.as_str().and_then(|text| text.parse().ok()))
})
.filter(|ttl| *ttl != 0)
.or_else(|| request.get("ttl").and_then(Value::as_i64))
.unwrap_or(900);
object.insert("ttl".into(), json!(ttl));
}
run_dns_cli(&dns_cli_args(&restore, "upsert"))?;
return Ok("restored the prior DNS record".to_string());
}
if let Some(current) = current.as_ref()
&& current.get("value") == request.get("record_value")
{
run_dns_cli(&dns_cli_args(request, "delete"))?;
return Ok("removed the partially created DNS record".to_string());
}
Ok("no DNS compensation was needed".to_string())
};
match attempt() {
Ok(detail) => (true, detail),
Err(error) => (false, error.to_string()),
}
}
fn execute_dns(store: &OperationStore, op_id: &str) -> Result<Value> {
let (record, fence) = set_running(store, op_id)?;
if record.get("state").and_then(Value::as_str) == Some("completed") {
return Ok(record);
}
let request = record.get("request").cloned().unwrap_or(json!({}));
let mut prior: Option<Value> = None;
let outcome = (|| -> Result<Value> {
let (_, found) = dns_lookup(&request)?;
prior = found.clone();
let recorded = prior.clone().unwrap_or(Value::Null);
update_for_fence(store, op_id, fence, |current| {
if let Some(object) = current.as_object_mut() {
object.insert("prior_dns_record".into(), recorded);
}
Ok(())
})?;
run_dns_cli(&dns_cli_args(&request, "upsert"))?;
let (source, created) = dns_lookup(&request)?;
let created = match created {
Some(created) if created.get("value") == request.get("record_value") => created,
_ => bail!("DNS upsert returned without the requested record"),
};
update_for_fence(store, op_id, fence, |current| {
if let Some(object) = current.as_object_mut() {
object.insert("state".into(), json!("completed"));
object.insert("lease_expires_at".into(), Value::Null);
}
if let Some(step) = current
.get_mut("steps")
.and_then(Value::as_array_mut)
.and_then(|steps| steps.first_mut())
.and_then(Value::as_object_mut)
{
step.insert("state".into(), json!("completed"));
step.remove("reason");
}
let record_id = match created.get("id") {
Some(Value::String(text)) => text.clone(),
Some(other) => other.to_string(),
None => "None".to_string(),
};
if let Some(object) = current.as_object_mut() {
object.insert(
"resource_ids".into(),
json!({
"dns": {
"provider": source.get("provider").cloned().unwrap_or(Value::Null),
"zone": source.get("zone").cloned().unwrap_or(Value::Null),
"record_id": record_id,
"fqdn": created.get("fqdn").cloned().unwrap_or(Value::Null),
"type": created.get("type").cloned().unwrap_or(Value::Null),
}
}),
);
}
append_audit(
current,
"dns_handoff_completed",
vec![("fence", json!(fence))],
);
Ok(())
})
})();
match outcome {
Ok(record) => Ok(record),
Err(error) => {
let (compensated, detail) = compensate_dns(&request, prior.as_ref());
let message = error.to_string();
update_for_fence(store, op_id, fence, |current| {
let step_state = if compensated {
"rolled_back"
} else {
"compensation_pending"
};
if let Some(object) = current.as_object_mut() {
object.insert("lease_expires_at".into(), Value::Null);
}
if let Some(step) = current
.get_mut("steps")
.and_then(Value::as_array_mut)
.and_then(|steps| steps.first_mut())
.and_then(Value::as_object_mut)
{
step.insert("state".into(), json!(step_state));
step.insert("reason".into(), json!(message));
}
if let Some(object) = current.as_object_mut() {
object.insert("state".into(), json!(step_state));
}
append_audit(
current,
"dns_handoff_failed",
vec![
("fence", json!(fence)),
("error", json!(message)),
("compensation", json!(detail)),
],
);
Ok(())
})
}
}
}
fn workspace_usage() -> ! {
eprintln!("usage: akurai-ec2 workspace [-h] {{request,resume,approve-delete,status,list}} ...");
std::process::exit(2);
}
fn take_value(args: &[String], index: &mut usize, flag: &str) -> String {
*index += 1;
match args.get(*index) {
Some(value) => value.clone(),
None => {
eprintln!("akurai-ec2 workspace: error: argument {flag}: expected one argument");
std::process::exit(2);
}
}
}
fn workspace_dispatch(args: &[String]) -> Result<Value> {
let store = OperationStore::new()?;
let Some(command) = args.first().map(String::as_str) else {
workspace_usage();
};
match command {
"request" => {
let mut parsed = RequestArgs {
provider: "ec2".to_string(),
ttl: 900,
..RequestArgs::default()
};
let mut index = 1;
let mut action: Option<String> = None;
while index < args.len() {
match args[index].as_str() {
"--idempotency-key" => {
parsed.idempotency_key = take_value(args, &mut index, "--idempotency-key")
}
"--organization" => {
parsed.organization = take_value(args, &mut index, "--organization")
}
"--workspace" => parsed.workspace = take_value(args, &mut index, "--workspace"),
"--domain" => parsed.domain = Some(take_value(args, &mut index, "--domain")),
"--record-type" => {
parsed.record_type = Some(take_value(args, &mut index, "--record-type"))
}
"--record-value" => {
parsed.record_value = Some(take_value(args, &mut index, "--record-value"))
}
"--zone" => parsed.zone = Some(take_value(args, &mut index, "--zone")),
"--provider" => {
let value = take_value(args, &mut index, "--provider");
if value != "ec2" && value != "1984" {
eprintln!(
"akurai-ec2 workspace: error: argument --provider: invalid choice: '{value}' (choose from 'ec2', '1984')"
);
std::process::exit(2);
}
parsed.provider = value;
}
"--ttl" => {
let value = take_value(args, &mut index, "--ttl");
match value.parse::<i64>() {
Ok(ttl) => parsed.ttl = ttl,
Err(_) => {
eprintln!(
"akurai-ec2 workspace: error: argument --ttl: invalid int value: '{value}'"
);
std::process::exit(2);
}
}
}
"--irreversible" => parsed.irreversible = true,
"--apply" => parsed.apply = true,
other if other.starts_with("--") => workspace_usage(),
other => {
if action.is_some() {
workspace_usage();
}
if !WORKSPACE_ACTIONS.contains(&other) {
eprintln!(
"akurai-ec2 workspace: error: argument action: invalid choice: '{other}'"
);
std::process::exit(2);
}
action = Some(other.to_string());
}
}
index += 1;
}
let Some(action) = action else {
eprintln!(
"akurai-ec2 workspace: error: the following arguments are required: action"
);
std::process::exit(2);
};
if parsed.idempotency_key.is_empty()
|| parsed.organization.is_empty()
|| parsed.workspace.is_empty()
{
eprintln!(
"akurai-ec2 workspace: error: the following arguments are required: --idempotency-key, --organization, --workspace"
);
std::process::exit(2);
}
parsed.action = action;
create_or_get(&store, &parsed)
}
"resume" => {
let mut op_id: Option<String> = None;
let mut apply = false;
for argument in &args[1..] {
match argument.as_str() {
"--apply" => apply = true,
other if other.starts_with("--") => workspace_usage(),
other => op_id = Some(other.to_string()),
}
}
let Some(op_id) = op_id else {
eprintln!(
"akurai-ec2 workspace: error: the following arguments are required: operation_id"
);
std::process::exit(2);
};
resume(&store, &op_id, apply)
}
"approve-delete" => {
let mut op_id: Option<String> = None;
let mut token: Option<String> = None;
let mut index = 1;
while index < args.len() {
match args[index].as_str() {
"--approval-token" => {
token = Some(take_value(args, &mut index, "--approval-token"))
}
other if other.starts_with("--") => workspace_usage(),
other => op_id = Some(other.to_string()),
}
index += 1;
}
let (Some(op_id), Some(token)) = (op_id, token) else {
eprintln!(
"akurai-ec2 workspace: error: the following arguments are required: operation_id, --approval-token"
);
std::process::exit(2);
};
approve_delete(&store, &op_id, &token)
}
"status" => {
let Some(op_id) = args.get(1) else {
eprintln!(
"akurai-ec2 workspace: error: the following arguments are required: operation_id"
);
std::process::exit(2);
};
store.read(|state| Ok(response(find_operation(state, op_id)?)))
}
"list" => store.read(|state| {
let operations = state
.get("operations")
.and_then(Value::as_object)
.ok_or_else(|| anyhow!("workspace lifecycle journal has an unsupported schema"))?;
Ok(Value::Array(
operations.values().map(response).collect::<Vec<Value>>(),
))
}),
_ => workspace_usage(),
}
}
/// Durable, fenced workspace lifecycle control. Unsupported source operations
/// stay explicitly pending; DNS is the one applied mutation.
pub fn workspace(_ec2: &Ec2, args: &[String]) -> Result<()> {
match workspace_dispatch(args) {
Ok(result) => {
println!("{}", compact_json(&result));
Ok(())
}
Err(error) => {
let message = serde_json::to_string(&error.to_string())
.unwrap_or_else(|_| "\"lifecycle error\"".to_string());
eprintln!("{{\"ok\":false,\"error\":{}}}", ascii_escaped(&message));
std::process::exit(1);
}
}
}
// ===========================================================================
// backups and monitoring
// ===========================================================================
fn backup_dispatch(
kind: &str,
action: Option<&str>,
retention: Option<u32>,
default_retention: u32,
maximum: u32,
) -> Result<(String, u32)> {
let action = action.unwrap_or("verify").to_string();
if !matches!(action.as_str(), "install" | "run" | "verify") {
bail!("{kind} backup action must be install, run, or verify");
}
let retention = retention.unwrap_or(default_retention);
if !(2..=maximum).contains(&retention) {
bail!("{kind} backup retention must be an integer from 2 through {maximum}");
}
Ok((action, retention))
}
const BUNFORK_BACKUP_INSTALL: &str = include_str!("remote/bunfork-backup-install.sh");
const BUNFORK_BACKUP_VERIFY: &str = include_str!("remote/bunfork-backup-verify.sh");
const TASKS_BACKUP_INSTALL: &str = include_str!("remote/tasks-backup-install.sh");
const TASKS_BACKUP_VERIFY: &str = include_str!("remote/tasks-backup-verify.sh");
const WORKBENCH_KEY_INSTALL: &str = include_str!("remote/workbench-key-install.sh");
const WORKBENCH_UPLOAD: &str = include_str!("remote/workbench-upload.sh");
const WORKBENCH_VERIFY: &str = include_str!("remote/workbench-verify.sh");
const MONITOR_INSTALL: &str = include_str!("remote/bunfork-monitor-install.sh");
const MONITOR_VERIFY: &str = include_str!("remote/bunfork-monitor-verify.sh");
const MAIL_BACKUP_INSTALL: &str = include_str!("remote/mail-backup-install.sh");
const MAIL_BACKUP_VERIFY: &str = include_str!("remote/mail-backup-verify.sh");
const MAIL_AUDIT: &str = include_str!("remote/mail-audit.sh");
const MAIL_UPDATE: &str = include_str!("remote/mail-update.sh");
const MAIL_DOMAIN: &str = include_str!("remote/mail-domain.sh");
/// The mail server pushes its encrypted archives to Titan over the mesh overlay,
/// and only Titan holds the secret half of the recipient key.
const MAIL_OFFSITE_DEST: &str = "olibuijr@100.88.0.9";
const MAIL_OFFSITE_RECIPIENT: &str = "akurai-mail-offsite@olibuijr.com";
const MAIL_HOST: &str = "mail.olibuijr.com";
/// Install, run, or verify the encrypted Bunfork database backup timer.
pub fn bunfork_backup(ec2: &Ec2, action: Option<&str>, retention: Option<u32>) -> Result<()> {
let (action, retention) = backup_dispatch("bunfork", action, retention, 14, 365)?;
if action == "install" {
say("installing encrypted Bunfork backup service and timer");
ec2.ssh_stdin_streaming(
&format!("bash -s -- '{retention}'"),
BUNFORK_BACKUP_INSTALL.as_bytes(),
)?;
} else if action == "run" {
say("running an encrypted Bunfork backup");
ec2.ssh_streaming("sudo systemctl start bunfork-backup.service")?;
}
say("verifying encrypted Bunfork backup and recovery path");
ec2.ssh_stdin_streaming("bash -s", BUNFORK_BACKUP_VERIFY.as_bytes())
}
/// Install, run, or verify the hourly encrypted AkurAI-Tasks backup.
pub fn tasks_backup(ec2: &Ec2, action: Option<&str>, retention: Option<u32>) -> Result<()> {
let (action, retention) = backup_dispatch("AkurAI-Tasks", action, retention, 48, 720)?;
if action == "install" {
say("installing encrypted AkurAI-Tasks backup service and hourly timer");
ec2.ssh_stdin_streaming(
&format!("bash -s -- '{retention}'"),
TASKS_BACKUP_INSTALL.as_bytes(),
)?;
} else if action == "run" {
say("running an encrypted AkurAI-Tasks backup");
ec2.ssh_streaming("sudo systemctl start akurai-tasks-backup.service")?;
}
say("verifying encrypted AkurAI-Tasks backup and isolated recovery");
ec2.ssh_stdin_streaming("bash -s", TASKS_BACKUP_VERIFY.as_bytes())
}
fn gpg_fingerprint(homedir: &Path, recipient: &str) -> Result<String> {
let Some(home_text) = homedir.to_str() else {
bail!("invalid gnupg home path");
};
let listing = capture_local(
Path::new("gpg"),
&[
"--batch",
"--homedir",
home_text,
"--with-colons",
"--fingerprint",
recipient,
],
"gpg --fingerprint",
)?;
let fingerprint = listing
.lines()
.find(|line| line.starts_with("fpr:"))
.and_then(|line| line.split(':').nth(9))
.unwrap_or_default()
.to_string();
Ok(fingerprint)
}
fn valid_fingerprint(value: &str) -> bool {
value.len() == 40
&& value
.chars()
.all(|c| c.is_ascii_digit() || ('A'..='F').contains(&c))
}
fn first_bytes_hex(path: &Path, count: usize) -> Result<String> {
let mut file = File::open(path).with_context(|| format!("read {}", path.display()))?;
let mut buffer = vec![0u8; count];
let mut filled = 0;
while filled < count {
let read = file.read(&mut buffer[filled..])?;
if read == 0 {
break;
}
filled += read;
}
Ok(buffer[..filled]
.iter()
.map(|byte| format!("{byte:02x}"))
.collect())
}
/// Encrypt Workbench snapshots on this machine and store them on EC2.
pub fn workbench_backup(ec2: &Ec2, action: Option<&str>, retention: Option<u32>) -> Result<()> {
let (action, retention) = backup_dispatch("Workbench", action, retention, 14, 365)?;
let root = home()?;
let workbench_ctl = match env::var_os("OMP_WORKBENCH_CTL") {
Some(value) if !value.is_empty() => PathBuf::from(value),
_ => root.join(".local/bin/titan-omp-workbenchctl"),
};
let offsite_home = match env::var_os("OMP_WORKBENCH_OFFSITE_HOME") {
Some(value) if !value.is_empty() => PathBuf::from(value),
_ => root.join(".local/share/omp-workbench/offsite"),
};
let offsite_gnupg = offsite_home.join("gnupg");
let fingerprint_file = offsite_home.join("recipient.fingerprint");
let recipient = "titan-omp-workbench-offsite@olibuijr.com";
let service_path = root.join(".config/systemd/user/omp-workbench-offsite-backup.service");
let timer_path = root.join(".config/systemd/user/omp-workbench-offsite-backup.timer");
if action == "install" {
say("installing EC2 encryption identity and backup destination");
let public_key = ec2.ssh_stdin(
&format!("bash -s -- {}", shell_quote(recipient)),
WORKBENCH_KEY_INSTALL.as_bytes(),
)?;
if !public_key.contains("BEGIN PGP PUBLIC KEY BLOCK") {
bail!("EC2 did not return a valid backup public key");
}
make_private_dir(&offsite_home, 0o700)?;
make_private_dir(&offsite_gnupg, 0o700)?;
let public_key_file = offsite_home.join("recipient.asc");
write_private(&public_key_file, &format!("{}\n", public_key.trim_end()))?;
let Some(gnupg_text) = offsite_gnupg.to_str() else {
bail!("invalid offsite gnupg path");
};
let Some(key_text) = public_key_file.to_str() else {
bail!("invalid public key path");
};
if !local_ok(
"gpg",
&["--batch", "--homedir", gnupg_text, "--import", key_text],
) {
bail!("failed to import EC2 backup public key");
}
let fingerprint = gpg_fingerprint(&offsite_gnupg, recipient)?;
if !valid_fingerprint(&fingerprint) {
bail!("invalid EC2 backup key fingerprint");
}
let mut ownertrust = Command::new("gpg")
.args(["--batch", "--homedir", gnupg_text, "--import-ownertrust"])
.stdin(Stdio::piped())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.context("failed to start gpg --import-ownertrust")?;
if let Some(stdin) = ownertrust.stdin.as_mut() {
stdin.write_all(format!("{fingerprint}:6:\n").as_bytes())?;
}
let status = ownertrust
.wait()
.context("failed to wait for gpg --import-ownertrust")?;
if !status.success() {
bail!("failed to import EC2 backup key ownertrust");
}
write_private(&fingerprint_file, &format!("{fingerprint}\n"))?;
let unit_dir = service_path
.parent()
.ok_or_else(|| anyhow!("no systemd user unit directory"))?
.to_path_buf();
make_private_dir(&unit_dir, 0o700)?;
let exe = env::current_exe().unwrap_or_else(|_| PathBuf::from("akurai"));
let service = format!(
"[Unit]\nDescription=Create and send an encrypted Workbench backup to AkurAI EC2\nAfter=network-online.target omp-workbench.service\nWants=network-online.target\n\n[Service]\nType=oneshot\nExecStart={} ec2 workbench-backup run {retention}\nUMask=0077\nNoNewPrivileges=true\n\n[Install]\nWantedBy=default.target\n",
exe.display()
);
let timer = "[Unit]\nDescription=Daily off-machine Workbench backup\n\n[Timer]\nOnCalendar=*-*-* 04:15:00 UTC\nRandomizedDelaySec=15m\nPersistent=true\nUnit=omp-workbench-offsite-backup.service\n\n[Install]\nWantedBy=timers.target\n";
write_private(&service_path, &service)?;
write_private(&timer_path, timer)?;
run_local(
Path::new("systemctl"),
&["--user", "daemon-reload"],
"systemctl --user daemon-reload",
)?;
run_local(
Path::new("systemctl"),
&[
"--user",
"enable",
"--now",
"omp-workbench-offsite-backup.timer",
],
"systemctl --user enable",
)?;
run_local(
Path::new("systemctl"),
&["--user", "start", "omp-workbench-offsite-backup.service"],
"systemctl --user start",
)?;
} else if action == "run" {
if !workbench_ctl.is_file() {
bail!(
"Workbench controller not found: {}",
workbench_ctl.display()
);
}
if !offsite_gnupg.is_dir() {
bail!("offsite encryption identity is not installed");
}
say("creating and encrypting a Workbench snapshot");
let snapshot = capture_local(&workbench_ctl, &["backup"], "workbench backup")?;
let expected_prefix = root.join(".local/share/omp-workbench/backups/workbench-");
let matches_shape = expected_prefix
.to_str()
.map(|prefix| snapshot.starts_with(prefix) && snapshot.ends_with(".sqlite"))
.unwrap_or(false);
if !matches_shape {
bail!("Workbench controller returned an unexpected snapshot path");
}
let fingerprint = fs::read_to_string(&fingerprint_file)
.map_err(|_| anyhow!("backup recipient fingerprint is not installed"))?
.trim()
.to_string();
if !valid_fingerprint(&fingerprint) {
bail!("invalid backup recipient fingerprint");
}
let ciphertext = tempfile::Builder::new()
.prefix(".outgoing-")
.suffix(".sqlite.gpg")
.tempfile_in(&offsite_home)
.context("create the outgoing ciphertext")?;
let ciphertext_path = ciphertext.path().to_path_buf();
let (Some(gnupg_text), Some(cipher_text)) =
(offsite_gnupg.to_str(), ciphertext_path.to_str())
else {
bail!("invalid offsite paths");
};
run_local(
Path::new("gpg"),
&[
"--batch",
"--yes",
"--homedir",
gnupg_text,
"--recipient",
&format!("{fingerprint}!"),
"--output",
cipher_text,
"--encrypt",
&snapshot,
],
"gpg --encrypt",
)?;
if fs::metadata(&ciphertext_path)
.map(|meta| meta.len())
.unwrap_or(0)
== 0
{
bail!("encrypted backup is empty");
}
if first_bytes_hex(&ciphertext_path, 16)? == "53514c69746520666f726d6174203300" {
bail!("backup was not encrypted");
}
let name = format!(
"workbench-{}-{:09}.sqlite.gpg",
compact_stamp(epoch_now()),
nanos_now()
);
let remote_tmp = ec2
.ssh(
r#"find /tmp -maxdepth 1 -user "$USER" -type f -name "omp-workbench-backup-*.gpg" -mmin +1440 -delete; mktemp /tmp/omp-workbench-backup-XXXXXXXX.gpg"#,
)?
.trim()
.to_string();
ec2.ship(&ciphertext_path, &remote_tmp)?;
let upload = ec2.ssh_stdin_streaming(
&format!(
"bash -s -- {} {} {retention}",
shell_quote(&remote_tmp),
shell_quote(&name)
),
WORKBENCH_UPLOAD.as_bytes(),
);
let _ = fs::remove_file(&ciphertext_path);
if upload.is_err() {
let _ = ec2.ssh(&format!("rm -f -- {}", shell_quote(&remote_tmp)));
upload?;
}
}
say("verifying off-machine Workbench backup and isolated restore");
if action != "run" {
run_local(
Path::new("systemctl"),
&[
"--user",
"is-enabled",
"--quiet",
"omp-workbench-offsite-backup.timer",
],
"systemctl --user is-enabled",
)?;
run_local(
Path::new("systemctl"),
&[
"--user",
"is-active",
"--quiet",
"omp-workbench-offsite-backup.timer",
],
"systemctl --user is-active",
)?;
let result = capture_local(
Path::new("systemctl"),
&[
"--user",
"show",
"omp-workbench-offsite-backup.service",
"-p",
"Result",
"--value",
],
"systemctl --user show",
)?;
if result.trim() != "success" {
bail!("omp-workbench-offsite-backup.service last result is not success");
}
}
ec2.ssh_stdin_streaming(
&format!("bash -s -- {retention}"),
WORKBENCH_VERIFY.as_bytes(),
)
}
struct MailOffsite {
home: PathBuf,
gnupg: PathBuf,
archives: PathBuf,
}
fn mail_offsite() -> Result<MailOffsite> {
let root = home()?;
let home_dir = match env::var_os("AKURAI_MAIL_OFFSITE_HOME") {
Some(value) if !value.is_empty() => PathBuf::from(value),
_ => root.join(".local/share/akurai-mail-offsite"),
};
let archives = match env::var_os("AKURAI_MAIL_ARCHIVE_DIR") {
Some(value) if !value.is_empty() => PathBuf::from(value),
_ => root.join("backups/akurai-mail"),
};
Ok(MailOffsite {
gnupg: home_dir.join("gnupg"),
home: home_dir,
archives,
})
}
fn file_sha256(path: &Path) -> Result<String> {
let mut file = File::open(path).with_context(|| format!("read {}", path.display()))?;
let mut hasher = Sha256::new();
let mut buffer = vec![0u8; 65_536];
loop {
let read = file.read(&mut buffer)?;
if read == 0 {
break;
}
hasher.update(&buffer[..read]);
}
Ok(hasher
.finalize()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect())
}
/// Titan's archives, newest first.
fn mail_archives(offsite: &MailOffsite) -> Result<Vec<(SystemTime, PathBuf)>> {
let mut archives: Vec<(SystemTime, PathBuf)> = fs::read_dir(&offsite.archives)
.with_context(|| format!("read {}", offsite.archives.display()))?
.flatten()
.filter(|entry| {
let name = entry.file_name().to_string_lossy().into_owned();
name.starts_with("akurai-mail-") && name.ends_with(".tar.gz.gpg")
})
.filter_map(|entry| {
let modified = entry.metadata().ok()?.modified().ok()?;
Some((modified, entry.path()))
})
.collect();
if archives.is_empty() {
bail!(
"no encrypted mail archive in {}",
offsite.archives.display()
);
}
archives.sort_by_key(|entry| std::cmp::Reverse(entry.0));
Ok(archives)
}
/// Decrypt one archive into a scratch directory and list its members.
fn mail_archive_listing(
offsite: &MailOffsite,
archive: &Path,
) -> Result<(tempfile::TempDir, PathBuf, String)> {
let scratch = tempfile::Builder::new()
.prefix(".verify-")
.tempdir_in(&offsite.home)
.context("create the restore scratch directory")?;
let payload = scratch.path().join("payload.tar.gz");
let (Some(gnupg_text), Some(payload_text), Some(archive_text)) =
(offsite.gnupg.to_str(), payload.to_str(), archive.to_str())
else {
bail!("invalid offsite paths");
};
run_local(
Path::new("gpg"),
&[
"--batch",
"--yes",
"--homedir",
gnupg_text,
"--output",
payload_text,
"--decrypt",
archive_text,
],
"gpg --decrypt",
)?;
let listing = capture_local(Path::new("tar"), &["-tzf", payload_text], "tar --list")?;
Ok((scratch, payload, listing))
}
/// Restore the newest archive Titan holds, so `verify` proves recoverability
/// from this machine alone instead of trusting the mail server's own report.
fn verify_mail_offsite(offsite: &MailOffsite, retention: u32) -> Result<()> {
let archives = mail_archives(offsite)?;
let (modified, latest) = archives[0].clone();
if archives.len() > retention as usize {
bail!(
"{} holds {} archives, retention is {retention}",
offsite.archives.display(),
archives.len()
);
}
let age = SystemTime::now()
.duration_since(modified)
.map(|value| value.as_secs())
.unwrap_or(0);
if age >= 172_800 {
bail!("newest Titan mail archive is {age}s old");
}
let digest_file = PathBuf::from(format!("{}.sha256", latest.display()));
let recorded = fs::read_to_string(&digest_file)
.with_context(|| format!("read {}", digest_file.display()))?;
let recorded = recorded.split_whitespace().next().unwrap_or_default();
let actual = file_sha256(&latest)?;
if recorded != actual {
bail!("{} does not match its recorded sha256", latest.display());
}
let (_scratch, payload, listing) = mail_archive_listing(offsite, &latest)?;
let Some(payload_text) = payload.to_str() else {
bail!("invalid payload path");
};
for member in [
"state/manifest.json",
"var/vmail/",
"etc/postfix/main.cf",
"etc/dovecot/dovecot.conf",
] {
if !listing.lines().any(|line| line.starts_with(member)) {
bail!("the restored archive is missing {member}");
}
}
let manifest_text = capture_local(
Path::new("tar"),
&["-xzOf", payload_text, "state/manifest.json"],
"tar --extract manifest",
)?;
let manifest: Value =
serde_json::from_str(&manifest_text).context("parse the archive manifest")?;
let messages = manifest["messages"].as_i64().unwrap_or(0);
let bytes = manifest["vmail_bytes"].as_i64().unwrap_or(0);
let accounts = manifest["accounts"].as_i64().unwrap_or(0);
if messages <= 0 || bytes <= 0 {
bail!("the restored archive reports no mail");
}
println!(
"offsite={} latest={} age_secs={age} archives={} accounts={accounts} messages={messages} \
vmail_bytes={bytes} sha256=ok recovery=verified",
offsite.archives.display(),
latest
.file_name()
.map(|name| name.to_string_lossy().into_owned())
.unwrap_or_default(),
archives.len()
);
Ok(())
}
/// Install, run, or verify the encrypted mail-server backup Titan holds.
pub fn mail_backup(ec2: &Ec2, action: Option<&str>, retention: Option<u32>) -> Result<()> {
let (action, retention) = backup_dispatch("mail", action, retention, 14, 365)?;
let offsite = mail_offsite()?;
let fingerprint_file = offsite.home.join("recipient.fingerprint");
if action == "install" {
say("creating the Titan-held mail backup identity");
make_private_dir(&offsite.home, 0o700)?;
make_private_dir(&offsite.gnupg, 0o700)?;
make_private_dir(&offsite.archives, 0o700)?;
let Some(gnupg_text) = offsite.gnupg.to_str() else {
bail!("invalid offsite gnupg path");
};
let has_secret = capture_quiet(
"gpg",
&[
"--batch",
"--homedir",
gnupg_text,
"--with-colons",
"--list-secret-keys",
MAIL_OFFSITE_RECIPIENT,
],
)
.is_some_and(|listing| listing.lines().any(|line| line.starts_with("sec:")));
if !has_secret {
// Unattended restore intentionally uses an operator-only keyring
// without a passphrase; a stored passphrase adds no boundary here.
run_local(
Path::new("gpg"),
&[
"--batch",
"--homedir",
gnupg_text,
"--passphrase",
"",
"--quick-generate-key",
MAIL_OFFSITE_RECIPIENT,
"rsa3072",
"encrypt",
"0",
],
"gpg --quick-generate-key",
)?;
}
let fingerprint = gpg_fingerprint(&offsite.gnupg, MAIL_OFFSITE_RECIPIENT)?;
if !valid_fingerprint(&fingerprint) {
bail!("invalid mail backup key fingerprint");
}
write_private(&fingerprint_file, &format!("{fingerprint}\n"))?;
let public_key = capture_local(
Path::new("gpg"),
&[
"--batch",
"--homedir",
gnupg_text,
"--armor",
"--export",
MAIL_OFFSITE_RECIPIENT,
],
"gpg --export",
)?;
if !public_key.contains("BEGIN PGP PUBLIC KEY BLOCK") {
bail!("failed to export the mail backup public key");
}
let key_file = offsite.home.join("recipient.asc");
write_private(&key_file, &format!("{}\n", public_key.trim_end()))?;
say("installing the mail backup service and daily timer");
let Some(archive_text) = offsite.archives.to_str() else {
bail!("invalid mail archive directory");
};
let remote_key = ec2
.ssh("mktemp /tmp/akurai-mail-recipient-XXXXXXXX.asc")?
.trim()
.to_string();
ec2.ship(&key_file, &remote_key)?;
let install = ec2.ssh_stdin_streaming(
&format!(
"bash -s -- {} {} {retention} {} {}",
shell_quote(&remote_key),
shell_quote(&fingerprint),
shell_quote(MAIL_OFFSITE_DEST),
shell_quote(archive_text)
),
MAIL_BACKUP_INSTALL.as_bytes(),
);
if install.is_err() {
let _ = ec2.ssh(&format!("rm -f -- {}", shell_quote(&remote_key)));
install?;
}
} else if action == "run" {
say("running an encrypted mail backup");
ec2.ssh_streaming("sudo systemctl start akurai-mail-backup.service")?;
}
say("verifying the mail server's backup timer and encryption");
ec2.ssh_stdin_streaming(
&format!("bash -s -- {retention}"),
MAIL_BACKUP_VERIFY.as_bytes(),
)?;
say("restoring the newest Titan copy to prove recoverability");
verify_mail_offsite(&offsite, retention)
}
fn valid_mail_domain(value: &str) -> bool {
value.len() <= 253
&& value.contains('.')
&& !value.starts_with('.')
&& !value.ends_with('.')
&& value.split('.').all(|label| {
matches_class(label, |c| {
c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-'
}) && !label.starts_with('-')
&& !label.ends_with('-')
})
}
/// List the mail domains this box serves, or retire one that is decommissioned.
pub fn mail_domain(ec2: &Ec2, action: Option<&str>, domain: Option<&str>) -> Result<()> {
let action = action.unwrap_or("list");
if action == "list" {
say("mail domains on the mail server");
return ec2.ssh_stdin_streaming("bash -s -- list", MAIL_DOMAIN.as_bytes());
}
if action != "retire" {
bail!("mail domain action must be list or retire");
}
let Some(domain) = domain else {
bail!("mail domain retire needs a domain");
};
if !valid_mail_domain(domain) {
bail!("mail domain must be a lowercase dotted hostname");
}
// Retiring deletes the domain's mailbox store, so the offsite copy must
// already contain it: Titan is the only place it survives afterwards.
let offsite = mail_offsite()?;
let archives = mail_archives(&offsite)?;
let latest = archives[0].1.clone();
let (_scratch, _payload, listing) = mail_archive_listing(&offsite, &latest)?;
let prefix = format!("var/vmail/{domain}/");
let preserved = listing
.lines()
.filter(|line| line.starts_with(&prefix))
.count();
if preserved == 0 {
bail!(
"{} does not contain {prefix}: run `akurai-ec2 mail-backup run` before retiring {domain}",
latest.display()
);
}
println!(
"retire_gate=ok archive={} preserved_entries={preserved}",
latest
.file_name()
.map(|name| name.to_string_lossy().into_owned())
.unwrap_or_default()
);
say(&format!("retiring {domain} from the mail server"));
ec2.ssh_stdin_streaming(
&format!("bash -s -- retire {}", shell_quote(domain)),
MAIL_DOMAIN.as_bytes(),
)
}
fn valid_mail_host(value: &str) -> bool {
matches_class(value, |c| c.is_ascii_alphanumeric() || c == '.' || c == '-')
}
fn valid_mail_update_action(value: &str) -> bool {
matches!(value, "check" | "apply" | "reboot" | "postverify")
}
/// Audit the mail server's security posture. Read-only; exits non-zero on findings.
pub fn mail_audit(ec2: &Ec2, host: Option<&str>, warn_days: Option<u32>) -> Result<()> {
let host = host.unwrap_or(MAIL_HOST);
if !valid_mail_host(host) {
bail!("mail audit host must be a hostname");
}
let warn_days = warn_days.unwrap_or(21);
if !(1..=90).contains(&warn_days) {
bail!("mail audit warn days must be an integer from 1 through 90");
}
say(&format!("auditing mail server security posture for {host}"));
let code = ec2.ssh_stdin_streaming_status(
&format!("bash -s -- {} {warn_days}", shell_quote(host)),
MAIL_AUDIT.as_bytes(),
)?;
match code {
0 => Ok(()),
1 => bail!("mail audit reported findings: fix or escalate every FAIL line above"),
other => bail!("mail audit could not complete (exit {other})"),
}
}
/// Run one mail-update phase and translate its exit code into a real diagnosis.
fn mail_update_phase(ec2: &Ec2, action: &str) -> Result<()> {
let code =
ec2.ssh_stdin_streaming_status(&format!("bash -s -- {action}"), MAIL_UPDATE.as_bytes())?;
match code {
0 => Ok(()),
1 => bail!(
"mail update {action} reported a problem: read the refusal or the degraded \
unit/port/banner lines above"
),
other => bail!("mail update {action} could not complete (exit {other})"),
}
}
/// Report, install, or reboot into mail-server package updates.
pub fn mail_update(ec2: &Ec2, action: Option<&str>) -> Result<()> {
let action = action.unwrap_or("check");
if !valid_mail_update_action(action) {
bail!("mail update action must be check, apply, reboot, or postverify");
}
if action == "reboot" {
say("rebooting the mail server");
let _ = ec2.ssh_streaming_status("sudo systemctl reboot");
std::thread::sleep(Duration::from_secs(20));
let deadline = Instant::now() + Duration::from_secs(300);
while ec2.ssh("true").is_err() {
if Instant::now() >= deadline {
bail!("the mail server did not answer within 300s of the reboot");
}
std::thread::sleep(Duration::from_secs(10));
}
say("verifying mail services after the reboot");
return mail_update_phase(ec2, "postverify");
}
say(&format!("mail server package update: {action}"));
mail_update_phase(ec2, action)
}
fn valid_monitor_recipient(value: &str) -> bool {
let Some((local, domain)) = value.split_once('@') else {
return false;
};
matches_class(local, |c| {
c.is_ascii_alphanumeric() || ".!#$%&'*+/=?^_`{|}~-".contains(c)
}) && matches_class(domain, |c| {
c.is_ascii_alphanumeric() || c == '.' || c == '-'
})
}
/// Install, run, or verify Bunfork HTTPS readiness and TLS-expiry monitoring.
pub fn bunfork_monitor(
ec2: &Ec2,
action: Option<&str>,
domain: Option<&str>,
warn_days: Option<u32>,
recipient: Option<&str>,
) -> Result<()> {
let action = action.unwrap_or("verify");
let domain = domain.unwrap_or("bunfork.olibuijr.com");
let warn_days = warn_days.unwrap_or(30);
let recipient = recipient.unwrap_or("olibuijr@olibuijr.com");
if !matches!(action, "install" | "run" | "verify") {
bail!("monitor action must be install, run, or verify");
}
if !matches_class(domain, |c| {
c.is_ascii_alphanumeric() || c == '.' || c == '-'
}) {
bail!("invalid domain: {domain}");
}
if !(1..=3650).contains(&warn_days) {
bail!("TLS warning threshold must be from 1 through 3650 days");
}
if !valid_monitor_recipient(recipient) {
bail!("invalid alert recipient");
}
if action == "install" {
say("installing Bunfork readiness and TLS monitor");
ec2.ssh_stdin_streaming(
&format!("bash -s -- '{domain}' '{warn_days}' '{recipient}'"),
MONITOR_INSTALL.as_bytes(),
)?;
} else if action == "run" {
say("running Bunfork production monitor");
ec2.ssh_streaming("sudo systemctl start bunfork-monitor.service")?;
}
say("verifying Bunfork production monitor");
ec2.ssh_stdin_streaming(
&format!("bash -s -- '{domain}' '{warn_days}'"),
MONITOR_VERIFY.as_bytes(),
)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn secret_character_class_matches_the_shell_contract() {
let allowed = |value: &str| {
matches_class(value, |c| {
c.is_ascii_alphanumeric() || "._~!@%^*+=:,/-".contains(c)
})
};
assert!(allowed("abc123._~!@%^*+=:,/-"));
assert!(!allowed(""));
assert!(!allowed("has space"));
assert!(!allowed("dollar$sign"));
assert!(!allowed("back\\slash"));
}
#[test]
fn operation_id_is_stable_and_prefixed() {
let request = json!({"organization": "acme", "workspace": "core"});
let first = operation_id("provision", "key-1", &request);
let second = operation_id("provision", "key-1", &request);
assert_eq!(first, second);
assert!(first.starts_with("wop_"));
assert_eq!(first.len(), 28);
assert_ne!(first, operation_id("upgrade", "key-1", &request));
}
#[test]
fn iso_round_trips_through_the_journal_format() {
let stamp = iso_utc(1_767_225_600);
assert_eq!(stamp, "2026-01-01T00:00:00Z");
assert_eq!(parse_iso(&stamp).expect("parses"), 1_767_225_600);
assert_eq!(compact_stamp(1_767_225_600), "20260101T000000");
}
#[test]
fn json_is_ascii_escaped_like_python() {
assert_eq!(compact_json(&json!({"a": "é"})), r#"{"a":"\u00e9"}"#);
}
#[test]
fn domain_validation_requires_a_fully_qualified_name() {
assert_eq!(
validate_workspace_domain("Example.COM.").expect("valid"),
"example.com"
);
assert!(validate_workspace_domain("localhost").is_err());
assert!(validate_workspace_domain("a..b.com").is_err());
}
#[test]
fn top_level_toml_keys_drop_comments_and_quotes() {
let keys = top_level_keys(
"# comment\nname = \"platform\"\nport = 8094 # inline\ndomain = \"a.example\"\n[section]\nport = 1\n",
);
assert_eq!(
keys.get("name").map(|(v, q)| (v.as_str(), *q)),
Some(("platform", true))
);
assert_eq!(
keys.get("port").map(|(v, q)| (v.as_str(), *q)),
Some(("8094", false))
);
assert_eq!(
keys.get("domain").map(|(v, _)| v.as_str()),
Some("a.example")
);
}
#[test]
fn retire_app_validators_match_the_python_patterns() {
assert!(valid_app_name("akurai-tasks"));
assert!(!valid_app_name("-leading"));
assert!(!valid_app_name("Upper"));
assert!(valid_retire_domain("tasks.olibuijr.com"));
assert!(!valid_retire_domain("-bad.example"));
}
#[test]
fn idempotency_keys_are_url_safe_and_bounded() {
assert!(valid_idempotency_key("Key.1:2_3-4"));
assert!(!valid_idempotency_key(""));
assert!(!valid_idempotency_key("-leading"));
assert!(!valid_idempotency_key(&"a".repeat(201)));
}
#[test]
fn mail_validators_bound_the_operator_surface() {
assert!(valid_mail_host("mail.olibuijr.com"));
assert!(!valid_mail_host("mail olibuijr.com"));
assert!(!valid_mail_host(""));
for action in ["check", "apply", "reboot", "postverify"] {
assert!(valid_mail_update_action(action));
}
assert!(!valid_mail_update_action("upgrade"));
let (action, retention) = backup_dispatch("mail", None, None, 14, 365).expect("defaults");
assert_eq!((action.as_str(), retention), ("verify", 14));
assert!(backup_dispatch("mail", Some("install"), Some(1), 14, 365).is_err());
assert!(backup_dispatch("mail", Some("install"), Some(366), 14, 365).is_err());
assert!(valid_mail_domain("example.com"));
assert!(valid_mail_domain("mail.olibuijr.com"));
assert!(!valid_mail_domain("localhost"));
assert!(!valid_mail_domain("Upper.Case"));
assert!(!valid_mail_domain("-bad.example"));
assert!(!valid_mail_domain("trailing.dot."));
assert!(!valid_mail_domain("semi;colon.example"));
}
#[test]
fn offsite_restore_refuses_an_empty_archive_directory() {
let scratch = tempfile::Builder::new()
.prefix("akurai-mail-offsite-")
.tempdir()
.expect("scratch");
let offsite = MailOffsite {
home: scratch.path().to_path_buf(),
gnupg: scratch.path().join("gnupg"),
archives: scratch.path().to_path_buf(),
};
let error = verify_mail_offsite(&offsite, 14).expect_err("no archive");
assert!(error.to_string().contains("no encrypted mail archive"));
}
}