diff --git a/daemon/Cargo.toml b/daemon/Cargo.toml new file mode 100644 index 0000000..66b738e --- /dev/null +++ b/daemon/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "smolboid" +version = "0.1.0" +edition = "2021" +description = "smolboid daemon: twtxt feeds published by smolmail" +license = "Apache-2.0" + +[dependencies] +smolboid-core = { path = "../core" } +fumi-core = { path = "../../fumi/core" } +serde = { version = "1", features = ["derive"] } +serde_json = "1" diff --git a/daemon/src/config.rs b/daemon/src/config.rs new file mode 100644 index 0000000..df96842 --- /dev/null +++ b/daemon/src/config.rs @@ -0,0 +1,102 @@ +//! Instance configuration: one flat `key = value` file, `#` comments. The +//! daemon's smolmail identity is bootstrapped out of band with the fumi CLI +//! against the same master key and store (`fumi trust`, `fumi register`). + +use std::path::{Path, PathBuf}; + +#[derive(Debug, Clone)] +pub struct Config { + /// HTTP listen address for the feed server. + pub listen: String, + /// Canonical origin the `# url` metadata carries. + pub base_url: String, + /// Host serving the daemon's smolmail mailbox. + pub smolmail_host: String, + /// Local part of the daemon's mailbox, for display. + pub mailbox: String, + /// 32-byte master secret, the fumi convention. + pub master: PathBuf, + /// State directory: state.json, mail.db and feeds/. + pub data: PathBuf, + pub posts_per_day: u32, + pub reposts_per_day: u32, + /// Per-feed size cap in bytes (PLAN.md, Retention). + pub cap_bytes: usize, + pub poll_secs: u64, + /// Network timeout for smolmail operations, seconds. + pub timeout: u64, +} + +impl Default for Config { + fn default() -> Config { + Config { + listen: "127.0.0.1:8080".into(), + base_url: "http://localhost:8080".into(), + smolmail_host: "localhost".into(), + mailbox: "twtxt".into(), + master: PathBuf::from("master.key"), + data: PathBuf::from("data"), + posts_per_day: 50, + reposts_per_day: 50, + cap_bytes: 1_048_576, + poll_secs: 60, + timeout: 30, + } + } +} + +impl Config { + pub fn load(path: &Path) -> Result { + let text = std::fs::read_to_string(path) + .map_err(|e| format!("cannot read {}: {e}", path.display()))?; + let mut cfg = Config::default(); + for (n, line) in text.lines().enumerate() { + let line = line.trim(); + if line.is_empty() || line.starts_with('#') { + continue; + } + let Some((key, value)) = line.split_once('=') else { + return Err(format!("line {}: expected `key = value`", n + 1)); + }; + let (key, value) = (key.trim(), value.trim()); + let bad = |_| format!("line {}: bad value for `{key}`", n + 1); + match key { + "listen" => cfg.listen = value.into(), + "base_url" => cfg.base_url = value.into(), + "smolmail_host" => cfg.smolmail_host = value.into(), + "mailbox" => cfg.mailbox = value.into(), + "master" => cfg.master = PathBuf::from(value), + "data" => cfg.data = PathBuf::from(value), + "posts_per_day" => cfg.posts_per_day = value.parse().map_err(&bad)?, + "reposts_per_day" => cfg.reposts_per_day = value.parse().map_err(&bad)?, + "cap_bytes" => cfg.cap_bytes = value.parse().map_err(&bad)?, + "poll_secs" => cfg.poll_secs = value.parse().map_err(&bad)?, + "timeout" => cfg.timeout = value.parse().map_err(&bad)?, + other => return Err(format!("line {}: unknown key `{other}`", n + 1)), + } + } + Ok(cfg) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_and_defaults() { + let cfg = Config::load(Path::new("no-such-file")).unwrap_err(); + assert!(cfg.contains("cannot read")); + + let text = "# instance\nlisten = 0.0.0.0:9000\nbase_url = https://smol.place\ncap_bytes = 4096\n"; + std::fs::write("/tmp/smolboid-test.toml", text).unwrap(); + let cfg = Config::load(Path::new("/tmp/smolboid-test.toml")).unwrap(); + assert_eq!(cfg.listen, "0.0.0.0:9000"); + assert_eq!(cfg.base_url, "https://smol.place"); + assert_eq!(cfg.cap_bytes, 4096); + assert_eq!(cfg.poll_secs, 60, "unset keys keep defaults"); + + std::fs::write("/tmp/smolboid-test.toml", "wat = 1\n").unwrap(); + assert!(Config::load(Path::new("/tmp/smolboid-test.toml")).is_err()); + } +} diff --git a/daemon/src/dispatch.rs b/daemon/src/dispatch.rs new file mode 100644 index 0000000..a061b02 --- /dev/null +++ b/daemon/src/dispatch.rs @@ -0,0 +1,290 @@ +//! Mail-to-action translation (PLAN.md, Management mail): a post mail is +//! bare text or a block whose only field is Reply-To; a command mail opens a +//! §5.5 block. Pure — the caller owns admission, retention and error replies. + +use smolboid_core::feed::MailLine; +use smolboid_core::rules::validate_text; +use smolboid_core::time; +use smolboid_core::{repost_body, reply_to_body}; + +use crate::state::PruneMode; + +/// One fetched mail, opened and verified by the caller. +pub struct Incoming { + pub sender: [u8; 32], + pub epk: [u8; 32], + pub sig: [u8; 64], + pub time: i64, + pub body: Vec, +} + +pub enum Action { + Post { + timestamp: String, + text: String, + /// The Reply-To value the body carried, when the ordinary-send block + /// form was used; bound in the feed's `smol:reply_to` metadata. + reply_to: Option, + mail: MailLine, + }, + Repost { + timestamp: String, + args: String, + reply_to: Option, + mail: MailLine, + }, + Commands(Vec), + /// Nothing to do: recognized by the rules as benign, never published. + Ignore, +} + +pub enum Command { + Nick(String), + Delete(String), + Prune(PruneMode), +} + +/// The `Twt-` frontmatter fields a command mail may carry; unknown keys are +/// preserved upstream and never alter behaviour (SPEC.md §5.5). +const F_NICK: &str = "twt-nick"; +const F_DELETE: &str = "twt-delete"; +const F_REPOST: &str = "twt-repost"; +const F_PRUNE: &str = "twt-prune"; +const F_REPLY_TO: &str = "reply-to"; + +/// Fails closed: a block with unreconstructible fields is never published. +pub fn dispatch(incoming: &Incoming) -> Result { + let text = std::str::from_utf8(&incoming.body).map_err(|_| "body is not UTF-8".to_owned())?; + + if !text.starts_with("---\n") { + // The bare post form: the body is the twt, byte for byte. + return post(incoming, text, None); + } + + let (fields, rest) = fumi_core::message::parse_frontmatter(text); + if fields.is_empty() { + // An empty or malformed block: nothing reliable, nothing published. + return Ok(Action::Ignore); + } + + let reply_to = fields.get(F_REPLY_TO).cloned(); + let beyond_reply_to: Vec<&String> = fields + .keys() + .filter(|k| k.as_str() != F_REPLY_TO) + .collect(); + + // A post in the ordinary-send form: a block whose only field is Reply-To. + if beyond_reply_to.is_empty() { + if let Some(reply_to) = &reply_to { + if text != reply_to_body(reply_to, &rest) { + return Err("post mail is not in canonical form".to_owned()); + } + return post(incoming, &rest, Some(reply_to.clone())); + } + } + + // A repost: the Twt-Repost field, optionally with the same Reply-To an + // ordinary send carries; both must reconstruct byte-exactly. + if fields.len() <= 2 { + if let Some(args) = fields.get(F_REPOST) { + let canonical_no = repost_body(args, None); + let canonical_yes = repost_body(args, reply_to.as_deref()); + if text == canonical_no || text == canonical_yes { + validate_text(args).map_err(|e| e.to_string())?; + return Ok(Action::Repost { + timestamp: time::render(incoming.time), + args: args.clone(), + reply_to, + mail: MailLine { + sender: incoming.sender, + epk: incoming.epk, + sig: incoming.sig, + }, + }); + } + return Err("repost mail is not in canonical form".to_owned()); + } + } + + // A block with extra fields and no Twt field: the author plainly meant to + // publish, and silence would hide the problem. + if !fields.contains_key(F_NICK) + && !fields.contains_key(F_DELETE) + && !fields.contains_key(F_PRUNE) + && !fields.contains_key(F_REPOST) + { + return Err( + "post mail carries fields beyond Reply-To (Subject, headers or an Accept token); \ + resend the bare text" + .to_owned(), + ); + } + + let mut commands = Vec::new(); + if let Some(nick) = fields.get(F_NICK) { + if !crate::feedstore::valid_username(nick) { + return Err("nick must be 1-30 bytes of a-z, 0-9 or hyphen".to_owned()); + } + commands.push(Command::Nick(nick.clone())); + } + if let Some(timestamp) = fields.get(F_DELETE) { + if time::parse(timestamp).is_none() { + return Err("Twt-Delete needs an RFC 3339 UTC timestamp".to_owned()); + } + commands.push(Command::Delete(timestamp.clone())); + } + if let Some(mode) = fields.get(F_PRUNE) { + commands.push(Command::Prune(match mode.as_str() { + "fifo" => PruneMode::Fifo, + "off" => PruneMode::Off, + other => return Err(format!("unknown prune mode `{other}`, want fifo or off")), + })); + } + // A mixed mail is a command mail; the body is ignored, never posted. + if commands.is_empty() { + Ok(Action::Ignore) + } else { + Ok(Action::Commands(commands)) + } +} + +fn post(incoming: &Incoming, text: &str, reply_to: Option) -> Result { + validate_text(text).map_err(|e| e.to_string())?; + Ok(Action::Post { + timestamp: time::render(incoming.time), + text: text.to_owned(), + reply_to, + mail: MailLine { + sender: incoming.sender, + epk: incoming.epk, + sig: incoming.sig, + }, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn incoming(body: &str) -> Incoming { + Incoming { + sender: [1; 32], + epk: [2; 32], + sig: [3; 64], + time: 1_760_000_000, + body: body.as_bytes().to_vec(), + } + } + + #[test] + fn plain_body_is_a_post() { + let Action::Post { timestamp, text, reply_to, mail } = dispatch(&incoming("hello smolverse")).unwrap() else { + panic!("expected a post"); + }; + assert_eq!(timestamp, "2025-10-09T08:53:20Z"); + assert_eq!(text, "hello smolverse"); + assert!(reply_to.is_none()); + assert_eq!(mail.sender, [1; 32]); + } + + #[test] + fn ordinary_send_form_is_a_post() { + // What `fumi send --body x` emits: a block with only Reply-To. + let body = "---\nReply-To: smol://tobias@smol.place/k\n---\nhello smolverse"; + let Action::Post { text, reply_to, .. } = dispatch(&incoming(body)).unwrap() else { + panic!("expected a post"); + }; + assert_eq!(text, "hello smolverse"); + assert_eq!(reply_to.as_deref(), Some("smol://tobias@smol.place/k")); + } + + #[test] + fn dashed_body_is_still_a_post() { + // A single line can never open a block, so nothing is escaped. + let Action::Post { text, .. } = dispatch(&incoming("--- dashed opening")).unwrap() else { + panic!("expected a post"); + }; + assert_eq!(text, "--- dashed opening"); + } + + #[test] + fn post_rules_are_enforced_with_reasons() { + assert!(dispatch(&incoming(&"a".repeat(161))).is_err()); + assert!(dispatch(&incoming("a\tb")).is_err()); + let bad = Incoming { body: vec![0xff, 0xfe], ..incoming("") }; + assert!(dispatch(&bad).is_err()); + } + + #[test] + fn extra_fields_make_a_rejected_post() { + // A Subject, or the Accept token of an accepted recipient. + assert!(dispatch(&incoming("---\nReply-To: smol://a/b\nSubject: hi\n---\nbody\n")).is_err()); + assert!(dispatch(&incoming("---\nAccept: token\n---\nbody\n")).is_err()); + } + + #[test] + fn canonical_repost_command_publishes() { + let args = "@ #abc1234"; + let no_reply_to = format!("---\nTwt-Repost: {args}\n---\n"); + let Action::Repost { args: got, reply_to, mail, .. } = + dispatch(&incoming(&no_reply_to)).unwrap() else { + panic!("expected a repost"); + }; + assert_eq!(got, args); + assert!(reply_to.is_none()); + assert_eq!(mail.sig, [3; 64]); + + // The ordinary-send form, Twt-Repost before Reply-To as fumi emits. + let with_reply_to = format!("---\nTwt-Repost: {args}\nReply-To: smol://tobias@s/k\n---\n"); + let Action::Repost { args: got, reply_to, .. } = + dispatch(&incoming(&with_reply_to)).unwrap() else { + panic!("expected a repost"); + }; + assert_eq!(got, args); + assert_eq!(reply_to.as_deref(), Some("smol://tobias@s/k")); + } + + #[test] + fn noncanonical_reposts_are_rejected() { + let args = "@ #abc"; + assert!(dispatch(&incoming(&format!("---\nTwt-Repost: {args} \n---\n"))).is_err()); + assert!(dispatch(&incoming(&format!("---\nTwt-Repost: {args}\n---\nextra\n"))).is_err()); + assert!(dispatch(&incoming(&format!( + "---\nTwt-Repost: {args}\nTwt-Nick: x\n---\n" + ))) + .is_err()); + // Reply-To after the closing marker, wrong order relative to the + // canonical forms. + assert!(dispatch(&incoming(&format!( + "---\nReply-To: smol://a/b\nTwt-Repost: {args}\n---\n" + ))) + .is_err()); + } + + #[test] + fn command_mails_apply_all_recognized_fields() { + let body = "---\nTwt-Nick: toby\nTwt-Prune: fifo\nSubject: ignored\n---\nignored body\n"; + let Action::Commands(cmds) = dispatch(&incoming(body)).unwrap() else { + panic!("expected commands"); + }; + assert_eq!(cmds.len(), 2); + assert!(matches!(cmds[0], Command::Nick(ref n) if n == "toby")); + assert!(matches!(cmds[1], Command::Prune(PruneMode::Fifo))); + } + + #[test] + fn command_values_are_validated() { + assert!(dispatch(&incoming("---\nTwt-Nick: Bad Nick\n---\n")).is_err()); + assert!(dispatch(&incoming("---\nTwt-Delete: not-a-date\n---\n")).is_err()); + assert!(dispatch(&incoming("---\nTwt-Prune: sometimes\n---\n")).is_err()); + } + + #[test] + fn odd_mail_shapes_fail_closed_to_ignore() { + // Malformed block, empty escape block, only unknown fields with a + // Twt- field absent — but pure unknown-field blocks are rejected + // as posts, so the ignore set is the malformed and empty ones. + assert!(matches!(dispatch(&incoming("---\nno colon here\n---\n")), Ok(Action::Ignore))); + assert!(matches!(dispatch(&incoming("---\n---\n")), Ok(Action::Ignore))); + } +} diff --git a/daemon/src/feedstore.rs b/daemon/src/feedstore.rs new file mode 100644 index 0000000..bb1874e --- /dev/null +++ b/daemon/src/feedstore.rs @@ -0,0 +1,129 @@ +//! The feed directory: one `twtxt.txt` per username under `/feeds/`. +//! The path is the verified smolmail username, so usernames are validated +//! before ever touching the filesystem. + +use std::io; +use std::path::{Path, PathBuf}; + +use smolboid_core::Feed; + +/// 1-30 bytes of `[a-z0-9-]`: a valid feed path segment and nick. +pub fn valid_username(name: &str) -> bool { + !name.is_empty() + && name.len() <= 30 + && name.bytes().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == b'-') +} + +#[derive(Clone)] +pub struct FeedStore { + root: PathBuf, +} + +impl FeedStore { + pub fn new(data: &Path) -> io::Result { + let root = data.join("feeds"); + std::fs::create_dir_all(&root)?; + Ok(FeedStore { root }) + } + + fn path(&self, username: &str) -> Result { + if !valid_username(username) { + return Err(format!("invalid username `{username}`")); + } + Ok(self.root.join(format!("{username}.txt"))) + } + + /// The feed file's mtime for `Last-Modified`; missing feeds are epoch. + pub fn path_mtime(&self, username: &str) -> Option { + self.path(username).ok().and_then(|p| p.metadata().ok()) + .and_then(|m| m.modified().ok()) + } + + pub fn exists(&self, username: &str) -> Result { + Ok(self.path(username)?.exists()) + } + + pub fn load(&self, username: &str) -> Result { + let path = self.path(username)?; + match std::fs::read_to_string(&path) { + Ok(text) => Ok(Feed::parse(&text)), + Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(Feed::new()), + Err(e) => Err(format!("cannot read {}: {e}", path.display())), + } + } + + /// Atomic replace, so a concurrent HTTP reader sees the old or the new + /// file, never a half-written one. + pub fn save(&self, username: &str, feed: &Feed) -> Result<(), String> { + let path = self.path(username)?; + let tmp = path.with_extension("tmp"); + std::fs::write(&tmp, feed.render()) + .map_err(|e| format!("cannot write {}: {e}", tmp.display()))?; + std::fs::rename(&tmp, &path) + .map_err(|e| format!("cannot replace {}: {e}", path.display())) + } + + /// Creates the feed skeleton at `admit` time: metadata only, no records. + pub fn create( + &self, + username: &str, + nick: &str, + address: &str, + base_url: &str, + host_key: &[u8; 32], + ) -> Result<(), String> { + if self.exists(username)? { + return Ok(()); + } + let mut feed = Feed::new(); + feed.set_meta("nick", nick); + feed.set_meta("url", &format!("{base_url}/twtxt/{username}.txt")); + feed.set_meta("smol:author", address); + feed.set_meta("smol:host_key", &fumi_core::crypto::b32(host_key)); + self.save(username, &feed) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn usernames_gate_the_filesystem_path() { + assert!(valid_username("tobias")); + assert!(valid_username("a-b2")); + assert!(!valid_username("")); + assert!(!valid_username("Tobias")); + assert!(!valid_username("a_b")); + assert!(!valid_username("../../etc/passwd")); + assert!(!valid_username(&"a".repeat(31))); + } + + #[test] + fn create_save_load_round_trip() { + let dir = std::env::temp_dir().join("smolboid-feeds-test"); + let _ = std::fs::remove_dir_all(&dir); + let feeds = FeedStore::new(&dir).unwrap(); + + feeds + .create("tobias", "tobias", "tobias@smol.place", "https://x", &[9u8; 32]) + .unwrap(); + assert!(feeds.exists("tobias").unwrap()); + assert!(!feeds.exists("dana").unwrap()); + + let feed = feeds.load("tobias").unwrap(); + assert_eq!(feed.author(), Some("tobias@smol.place")); + assert_eq!( + feed.meta("url"), + Some("https://x/twtxt/tobias.txt") + ); + + let mut feed = feeds.load("tobias").unwrap(); + feed.set_meta("nick", "toby"); + feeds.save("tobias", &feed).unwrap(); + assert_eq!(feeds.load("tobias").unwrap().meta("nick"), Some("toby")); + + // Path traversal is refused before any I/O. + assert!(feeds.load("../etc/passwd").is_err()); + } +} diff --git a/daemon/src/http.rs b/daemon/src/http.rs new file mode 100644 index 0000000..5b5a19b --- /dev/null +++ b/daemon/src/http.rs @@ -0,0 +1,315 @@ +//! The HTTP listener: one route, `GET /twtxt/.txt`, with the strong +//! ETag and caching behaviour the plan fixes. Hand-rolled on purpose — one +//! static file with conditional requests is smaller than any framework. + +use std::io::{Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::time::UNIX_EPOCH; + +use fumi_core::crypto::sha256; +use smolboid_core::time; + +use crate::feedstore::FeedStore; + +const MAX_REQUEST: usize = 8 * 1024; + +pub fn serve(listener: TcpListener, feeds: FeedStore) -> ! { + loop { + match listener.accept() { + Ok((stream, _)) => { + let feeds = feeds.clone(); + std::thread::spawn(move || handle(stream, feeds)); + } + Err(e) => eprintln!("http: accept failed: {e}"), + } + } +} + +fn handle(mut stream: TcpStream, feeds: FeedStore) { + // Bounded slow-client exposure: each connection gets its own thread, so + // a client that dribbles bytes must not hold it forever. + let _ = stream.set_read_timeout(Some(std::time::Duration::from_secs(10))); + let _ = stream.set_write_timeout(Some(std::time::Duration::from_secs(10))); + let request = match read_request(&mut stream) { + Some(r) => r, + None => return, + }; + let (method, path, headers) = match parse_request(&request) { + Some(parts) => parts, + None => { + let _ = write_simple(&mut stream, 400, "Bad Request", "bad request\n"); + return; + } + }; + if method != "GET" { + let _ = write_simple(&mut stream, 405, "Method Not Allowed", "GET only\n"); + return; + } + let Some(username) = username_of(path) else { + let _ = write_simple(&mut stream, 404, "Not Found", "not found\n"); + return; + }; + // An admitted author's feed exists from `admit` on; anything else, + // including a valid username with no feed, is not there to be served. + if !feeds.exists(&username).unwrap_or(false) { + let _ = write_simple(&mut stream, 404, "Not Found", "not found\n"); + return; + } + let Ok(feed) = feeds.load(&username) else { + let _ = write_simple(&mut stream, 404, "Not Found", "not found\n"); + return; + }; + let body = feed.render().into_bytes(); + let etag = hex(&sha256(&[&body])); + let modified = mtime(&feeds, &username); + let cached = matches_cache(&headers, &etag, modified); + + let mut head = format!( + "HTTP/1.1 {} {}\r\nETag: \"{etag}\"\r\nLast-Modified: {}\r\nCache-Control: max-age=300\r\nAccess-Control-Allow-Origin: *\r\n", + if cached { 304 } else { 200 }, + if cached { "Not Modified" } else { "OK" }, + http_date(modified), + ); + if cached { + head.push_str("\r\n"); + let _ = stream.write_all(head.as_bytes()); + return; + } + head.push_str(&format!( + "Content-Type: text/plain; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + )); + let _ = stream.write_all(head.as_bytes()); + let _ = stream.write_all(&body); +} + +/// `If-None-Match` takes precedence; `If-Modified-Since` applies only without +/// it (PLAN.md, Serving). +fn matches_cache(headers: &[(String, String)], etag: &str, modified: i64) -> bool { + if let Some((_, inm)) = headers.iter().find(|(k, _)| k == "if-none-match") { + return inm == "*" || inm.contains(&format!("\"{etag}\"")); + } + if let Some((_, ims)) = headers.iter().find(|(k, _)| k == "if-modified-since") { + if let Some(since) = parse_http_date(ims) { + return modified <= since; + } + } + false +} + +fn username_of(path: &str) -> Option { + let name = path.strip_prefix("/twtxt/")?.strip_suffix(".txt")?; + crate::feedstore::valid_username(name).then(|| name.to_owned()) +} + +fn read_request(stream: &mut TcpStream) -> Option> { + let mut buf = Vec::new(); + let mut chunk = [0u8; 512]; + loop { + let n = stream.read(&mut chunk).ok()?; + if n == 0 { + return None; + } + buf.extend_from_slice(&chunk[..n]); + if buf.windows(4).any(|w| w == b"\r\n\r\n") || buf.len() > MAX_REQUEST { + return Some(buf); + } + } +} + +fn parse_request(request: &[u8]) -> Option<(&str, &str, Vec<(String, String)>)> { + let text = std::str::from_utf8(request).ok()?; + let mut lines = text.split("\r\n"); + let request_line = lines.next()?; + let mut parts = request_line.split(' '); + let method = parts.next()?; + let path = parts.next()?; + let mut headers = Vec::new(); + for line in lines { + if line.is_empty() { + break; + } + if let Some((name, value)) = line.split_once(':') { + headers.push((name.trim().to_ascii_lowercase(), value.trim().to_owned())); + } + } + Some((method, path, headers)) +} + +fn write_simple(stream: &mut TcpStream, code: u16, label: &str, body: &str) -> std::io::Result<()> { + write!( + stream, + "HTTP/1.1 {code} {label}\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ) +} + +fn mtime(feeds: &FeedStore, username: &str) -> i64 { + feeds + .path_mtime(username) + .and_then(|t| t.duration_since(UNIX_EPOCH).ok()) + .map(|d| d.as_secs() as i64) + .unwrap_or(0) +} + +const DAYS: [&str; 7] = ["Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"]; +const MONTHS: [&str; 12] = [ + "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec", +]; + +/// IMF-fixdate, the only Last-Modified form (RFC 9110). +fn http_date(unix: i64) -> String { + let (y, mo, d, h, mi, s) = time::utc_parts(unix); + let wday = (unix.div_euclid(86_400) + 4).rem_euclid(7) as usize; // 1970-01-01: Thursday + format!( + "{}, {:02} {} {} {:02}:{:02}:{:02} GMT", + DAYS[wday], d, MONTHS[(mo - 1) as usize], y, h, mi, s + ) +} + +fn parse_http_date(text: &str) -> Option { + let rest = text.split(", ").nth(1)?; + let mut parts = rest.split(' '); + let date = parts.next()?; + let month = parts.next()?; + let year = parts.next()?; + let clock = parts.next()?; + if parts.next() != Some("GMT") { + return None; + } + let (d, mo) = month_of(date, month)?; + let mut clock = clock.split(':'); + let (h, mi, s) = ( + clock.next()?.parse().ok()?, + clock.next()?.parse().ok()?, + clock.next()?.parse().ok()?, + ); + time::from_parts(year.parse().ok()?, mo, d, h, mi, s) +} + +fn month_of(date: &str, month: &str) -> Option<(u32, u32)> { + let d: u32 = date.parse().ok()?; + let mo = MONTHS.iter().position(|m| *m == month)? as u32 + 1; + Some((d, mo)) +} + +fn hex(raw: &[u8]) -> String { + let mut out = String::with_capacity(raw.len() * 2); + for byte in raw { + out.push_str(&format!("{byte:02x}")); + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + fn dir(name: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!("smolboid-http-{name}")); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + + fn start(feeds: FeedStore) -> std::net::SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + std::thread::spawn(move || serve(listener, feeds)); + addr + } + + fn get(addr: &std::net::SocketAddr, request: &str) -> String { + use std::net::ToSocketAddrs; + let mut stream = TcpStream::connect(addr.to_socket_addrs().unwrap().next().unwrap()) + .expect("connect"); + stream.write_all(request.as_bytes()).unwrap(); + let mut response = Vec::new(); + stream.read_to_end(&mut response).unwrap(); + String::from_utf8_lossy(&response).into_owned() + } + + fn with_feed(name: &str) -> (FeedStore, String) { + let feeds = FeedStore::new(&dir(name)).unwrap(); + feeds.create("tobias", "tobias", "tobias@smol.place", "https://x", &[9; 32]).unwrap(); + let addr = start(feeds.clone()); + (feeds, format!("{addr}")) + } + + #[test] + fn serves_the_feed_with_the_planned_headers() { + let (_, addr) = with_feed("headers"); + let response = get(&addr.parse().unwrap(), "GET /twtxt/tobias.txt HTTP/1.1\r\nHost: x\r\n\r\n"); + assert!(response.starts_with("HTTP/1.1 200 OK\r\n"), "{response}"); + assert!(response.contains("ETag: \"")); + assert!(response.contains("Cache-Control: max-age=300")); + assert!(response.contains("Access-Control-Allow-Origin: *")); + assert!(response.contains("Content-Type: text/plain; charset=utf-8")); + assert!(response.contains("# smol:author = tobias@smol.place\n")); + assert!(response.ends_with(&format!( + "# smol:host_key = {}\n", + fumi_core::crypto::b32(&[9u8; 32]) + ))); + } + + #[test] + fn etag_match_answers_304() { + let (_, addr) = with_feed("etag"); + let addr: std::net::SocketAddr = addr.parse().unwrap(); + let first = get(&addr, "GET /twtxt/tobias.txt HTTP/1.1\r\n\r\n"); + let etag = first + .split("\r\n") + .find(|l| l.starts_with("ETag: ")) + .unwrap() + .trim_start_matches("ETag: "); + let hit = get( + &addr, + &format!("GET /twtxt/tobias.txt HTTP/1.1\r\nIf-None-Match: {etag}\r\n\r\n"), + ); + assert!(hit.starts_with("HTTP/1.1 304"), "{hit}"); + assert!(hit.contains("ETag: ")); + assert!(!hit.contains("ignored-body")); + } + + #[test] + fn if_none_match_takes_precedence_over_if_modified_since() { + let (_, addr) = with_feed("precedence"); + let addr: std::net::SocketAddr = addr.parse().unwrap(); + // An IMS after the file's mtime alone yields 304; a mismatching INM + // must win over that matching IMS. + let ahead = "Mon, 01 Jan 2035 00:00:00 GMT"; + let hit = get( + &addr, + &format!("GET /twtxt/tobias.txt HTTP/1.1\r\nIf-Modified-Since: {ahead}\r\n\r\n"), + ); + assert!(hit.starts_with("HTTP/1.1 304"), "{hit}"); + + let miss = get( + &addr, + &format!( + "GET /twtxt/tobias.txt HTTP/1.1\r\nIf-None-Match: \"deadbeef\"\r\nIf-Modified-Since: {ahead}\r\n\r\n" + ), + ); + assert!(miss.starts_with("HTTP/1.1 200"), "{miss}"); + } + + #[test] + fn unknown_paths_and_methods_do_not_publish() { + let (_, addr) = with_feed("paths"); + let addr: std::net::SocketAddr = addr.parse().unwrap(); + assert!(get(&addr, "GET /twtxt/../etc/passwd HTTP/1.1\r\n\r\n").starts_with("HTTP/1.1 404")); + assert!(get(&addr, "GET /nope.txt HTTP/1.1\r\n\r\n").starts_with("HTTP/1.1 404")); + // A valid username without a feed is not served as an empty feed. + assert!(get(&addr, "GET /twtxt/nobody.txt HTTP/1.1\r\n\r\n").starts_with("HTTP/1.1 404")); + assert!(get(&addr, "POST /twtxt/tobias.txt HTTP/1.1\r\n\r\n").starts_with("HTTP/1.1 405")); + } + + #[test] + fn http_dates_round_trip() { + for unix in [0i64, 1_760_000_000, 951_782_400] { + let date = http_date(unix); + assert_eq!(parse_http_date(&date), Some(unix), "{date}"); + } + assert_eq!(http_date(0), "Thu, 01 Jan 1970 00:00:00 GMT"); + } +} diff --git a/daemon/src/mailloop.rs b/daemon/src/mailloop.rs new file mode 100644 index 0000000..693c9de --- /dev/null +++ b/daemon/src/mailloop.rs @@ -0,0 +1,595 @@ +//! The mail loop: fetch from the daemon's mailbox, admit by ownership, apply +//! each mail to its feed, and send rate-limited error replies. The apply and +//! retention logic is pure and unit-tested; network glue stays thin. + +use fumi_core::account::Account; +use fumi_core::address::Address; +use fumi_core::client; +use fumi_core::store::{now, Store}; +use fumi_core::error::Error as FumiError; +use smolboid_core::feed::{Feed, MailLine, Record}; +use smolboid_core::envelope; + +use crate::config::Config; +use crate::dispatch::{self, Action, Command, Incoming}; +use crate::feedstore::FeedStore; +use crate::state::{payload_key, reply_allowed, FeedState, PruneMode, State}; + +use std::sync::{Mutex, MutexGuard}; + +/// Everything needed to publish one record, before retention gets a say. +pub struct PendingRecord { + pub timestamp: String, + pub text: String, + pub repost: bool, + /// The Reply-To the mail carried, when the ordinary-send form was used. + pub reply_to: Option, + pub mail: MailLine, +} + +/// Failures that draw an error reply (PLAN.md, Management mail). +#[derive(Debug)] +pub enum ApplyError { + OverQuota(bool), + Collision(String), + FeedFull, + /// The author's Reply-To changed after the feed bound one: published + /// records would stop reconstructing (PLAN.md, smol:reply_to). + ReplyToChanged, +} + +/// Applies a published record to the feed: quota, uniqueness, retention +/// (pruning in Fifo mode, rejecting in Off mode), then the append. +/// Returns the rendered size after appending, for the 90% warning. +pub fn apply_record( + feed: &mut Feed, + fs: &mut FeedState, + cfg: &Config, + record: PendingRecord, + now_unix: i64, +) -> Result<(), ApplyError> { + State::roll_counters(fs, now_unix); + let (counter, limit) = if record.repost { + (&mut fs.reposts, cfg.reposts_per_day) + } else { + (&mut fs.posts, cfg.posts_per_day) + }; + if *counter >= limit { + return Err(ApplyError::OverQuota(record.repost)); + } + if feed.has_timestamp(&record.timestamp) { + // The signature pins the time, so a bump is impossible; the author + // resends a second later (post rule 2). + return Err(ApplyError::Collision(record.timestamp)); + } + + // The Reply-To must be the author's own address (key = sender), and the + // verifier derives it from `smol:author` plus the record's sender key. The + // first publishing mail adopts the author's own spelling of the address; + // a later change cannot be adopted without invalidating old records. + if let Some(reply_to) = &record.reply_to { + let Ok(addr) = Address::parse(reply_to) else { + return Err(ApplyError::ReplyToChanged); + }; + if addr.identity != Some(record.mail.sender) || addr.short() != feed.author().unwrap_or_default() { + if feed.entries().is_empty() { + feed.set_meta("smol:author", &addr.short()); + } else { + return Err(ApplyError::ReplyToChanged); + } + } + } + + let entry = if record.repost { + Record::Repost { + timestamp: record.timestamp.clone(), + args: record.text.clone(), + } + } else { + Record::Post { + timestamp: record.timestamp.clone(), + text: record.text.clone(), + } + }; + let rendered = entry.render(); + let addition = rendered.len() + 1 + record.mail.render().len() + 1; + let mut size = feed.render().len() + addition; + if size > cfg.cap_bytes { + if fs.prune == PruneMode::Off { + return Err(ApplyError::FeedFull); + } + // FIFO: drop oldest records until the new one fits. If even an empty + // feed cannot hold it, the record simply exceeds the cap. + while size > cfg.cap_bytes { + let Some(oldest) = feed.entries().first().map(|e| e.record.timestamp().to_owned()) + else { + return Err(ApplyError::FeedFull); + }; + feed.delete(&oldest); + size = feed.render().len() + addition; + } + } + + feed.append(entry, Some(record.mail)); + *counter += 1; + + // The 90% warning fires once per growth episode (PLAN.md, Retention). + let size = feed.render().len(); + if size * 10 >= cfg.cap_bytes * 9 { + if !fs.warned { + fs.warned = true; + } + } else if fs.warned { + fs.warned = false; + } + Ok(()) +} + +/// Applies command fields to feed and state. Deletion failure is reported so +/// the caller can reply. +pub fn apply_commands( + feed: &mut Feed, + fs: &mut FeedState, + commands: &[Command], +) -> Result { + for command in commands { + match command { + Command::Nick(nick) => { + fs.nick = nick.clone(); + feed.set_meta("nick", nick); + } + Command::Delete(timestamp) => { + if !feed.delete(timestamp) { + return Err(format!("no record at {timestamp}")); + } + } + Command::Prune(mode) => fs.prune = *mode, + } + } + Ok(commands.len()) +} + +pub struct Daemon { + pub cfg: Config, + pub store: Store, + pub account: Account, + pub state: Mutex, + pub feeds: FeedStore, +} + +impl Daemon { + /// One poll cycle. Returns a human summary line. + pub fn poll(&self) -> Result { + let summary = client::fetch(&self.store, &self.account, false, false, self.cfg.timeout) + .map_err(|e| format!("fetch: {e}"))?; + if summary.stored == 0 { + return Ok("no new mail".into()); + } + let mut applied = 0; + let mut dropped = 0; + let mut replies: Vec<(String, String, String)> = Vec::new(); // (address, subject, body) + + let inbox = self + .store + .mail("inbox") + .map_err(|e| format!("inbox: {e}"))?; + for stored in &inbox { + let outcome = self.process(stored.id, &stored.envelope, &mut replies); + let _ = self.store.delete_local("inbox", &stored.id); + match outcome { + Ok(true) => applied += 1, + Ok(false) => dropped += 1, + Err(e) => eprintln!("mail: {e}"), + } + } + self.send_replies(replies)?; + Ok(format!( + "{} applied, {} dropped", + applied, dropped + )) + } + + /// Processes one stored envelope end to end. Ok(true) when the feed or + /// state changed. + fn process( + &self, + id: [u8; 32], + envelope: &[u8], + replies: &mut Vec<(String, String, String)>, + ) -> Result { + let opened = envelope::open(self.account.keys(), envelope, now()) + .map_err(|e| format!("open {}: {e}", &fumi_core::crypto::b32(&id)[..8]))?; + let incoming = Incoming { + sender: opened.sender, + epk: opened.epk, + sig: opened.sig, + time: opened.time, + body: opened.body, + }; + + let mut state = self.state.lock().unwrap(); + // Admission: a feed must exist and its owner must still be accepted. + // Requests-tier or blocked mail has neither and is dropped without a + // reply (PLAN.md, Admission and Error replies) — main-tier inbox rows + // are all `mail("inbox")` ever returns. + let index = match state.feed_of(&opened.sender) { + Some(index) => index, + None => match self.follow_rotation(&incoming, &mut state)? { + Some(index) => index, + None => return Ok(false), + }, + }; + let feed_state = state.feeds[index].clone(); + if !self.accepted(&feed_state.address)? { + return Ok(false); + } + + self.apply(&incoming, index, &mut state, replies) + } + + fn accepted(&self, address: &str) -> Result { + match self.store.accepted_state(address) { + Ok(Some(true)) => Ok(true), + Ok(_) => Ok(false), + Err(e) => Err(format!("accepted_state: {e}")), + } + } + + /// An unknown sender may be a rotated owner: a signed Reply-To names the + /// address, and a verified chain from the key we hold to this one + /// (SPEC.md §7) moves the feed to the new key, after which the very mail + /// that announced the rotation applies as the rotated owner. Anything + /// else is dropped. + fn follow_rotation( + &self, + incoming: &Incoming, + state: &mut MutexGuard<'_, State>, + ) -> Result, String> { + let Ok(text) = std::str::from_utf8(&incoming.body) else { + return Ok(None); + }; + let (fields, _) = fumi_core::message::parse_frontmatter(text); + let Some(reply_to) = fields.get("reply-to") else { + return Ok(None); + }; + let Ok(addr) = Address::parse(reply_to) else { + return Ok(None); + }; + let Some(user_key) = addr.identity else { + return Ok(None); + }; + if user_key != incoming.sender { + return Ok(None); + } + let Some(index) = state.feeds.iter().position(|f| f.address == addr.short()) else { + return Ok(None); + }; + let Some(old) = state.feeds[index].latest_owner() else { + return Ok(None); + }; + let mut session = client::connect(&self.store, &addr, false, self.cfg.timeout) + .map_err(|e| format!("rotation connect: {e}"))?; + let resolved = client::resolve(session.transport(), &addr.user) + .map_err(|e| format!("rotation resolve: {e}"))?; + session.close(); + let (current, chain) = resolved; + if !client::walk_chain(&addr.user, &old, ¤t, &chain) { + return Ok(None); + } + if state.feeds[index].owner_keys.last().map(String::as_str) + != Some(fumi_core::crypto::b32(¤t).as_str()) + { + state.feeds[index] + .owner_keys + .push(fumi_core::crypto::b32(¤t)); + } + Ok(Some(index)) + } + + fn apply( + &self, + incoming: &Incoming, + index: usize, + state: &mut MutexGuard<'_, State>, + replies: &mut Vec<(String, String, String)>, + ) -> Result { + let username = state.feeds[index].username.clone(); + let address = state.feeds[index].address.clone(); + let mut feed = self.feeds.load(&username)?; + let before = feed.render(); + + let outcome = match dispatch::dispatch(incoming) { + Ok(Action::Post { timestamp, text, reply_to, mail }) => { + self.publish( + state, index, &mut feed, &address, incoming, replies, + PendingRecord { timestamp, text, reply_to, repost: false, mail }, + )? + } + Ok(Action::Repost { timestamp, args, reply_to, mail }) => { + self.publish( + state, index, &mut feed, &address, incoming, replies, + PendingRecord { timestamp, text: args, reply_to, repost: true, mail }, + )? + } + Ok(Action::Commands(commands)) => { + match apply_commands(&mut feed, &mut state.feeds[index], &commands) { + Ok(n) => n > 0, + Err(reason) => { + self.queue_reply(state, incoming, &address, &reason, replies); + false + } + } + } + Ok(Action::Ignore) => false, + Err(reason) => { + self.queue_reply(state, incoming, &address, &reason, replies); + false + } + }; + + let after = feed.render(); + if after != before { + self.feeds.save(&username, &feed)?; + } + // Counters, ledgers and prune modes change even when no record did. + state + .save(&self.cfg.data.join("state.json")) + .map_err(|e| e.to_string())?; + Ok(outcome) + } + + #[allow(clippy::too_many_arguments)] + fn publish( + &self, + state: &mut MutexGuard<'_, State>, + index: usize, + feed: &mut Feed, + address: &str, + incoming: &Incoming, + replies: &mut Vec<(String, String, String)>, + record: PendingRecord, + ) -> Result { + let was_warned = state.feeds[index].warned; + match apply_record(feed, &mut state.feeds[index], &self.cfg, record, incoming.time) { + Ok(()) => { + // The transition false -> true sends the cap warning once per + // growth episode (PLAN.md, Retention). + if state.feeds[index].warned && !was_warned { + replies.push(( + address.to_owned(), + "twtxt: feed at 90% of its size cap".into(), + format!( + "Your feed has reached 90% of the {}-byte cap.\n\n\ + In hard-limit mode (the default) a full feed rejects new records; \ + delete some with Twt-Delete, or choose FIFO pruning with \ + Twt-Prune: fifo, which drops the oldest records to make room.", + self.cfg.cap_bytes + ), + )); + } + Ok(true) + } + Err(e) => { + let reason = match &e { + ApplyError::OverQuota(repost) => format!( + "daily {} quota exceeded; try tomorrow", + if *repost { "repost" } else { "post" } + ), + ApplyError::Collision(ts) => { + format!("timestamp {ts} already exists; wait a second and resend") + } + ApplyError::FeedFull => "feed is at its size cap; delete records with Twt-Delete or choose Twt-Prune: fifo".to_owned(), + ApplyError::ReplyToChanged => format!( + "your Reply-To does not match the feed's bound author {}; \ + send from that address or mail the operator", + feed.author().unwrap_or_default() + ), + }; + self.queue_reply(state, incoming, address, &reason, replies); + Ok(false) + } + } + } + + /// One error reply per rejected payload, within the hourly budget. + fn queue_reply( + &self, + state: &mut MutexGuard<'_, State>, + incoming: &Incoming, + address: &str, + reason: &str, + replies: &mut Vec<(String, String, String)>, + ) { + let key = payload_key(&incoming.sender, incoming.time, &incoming.body); + if !reply_allowed(&mut **state, &incoming.sender, &key, now()) { + return; + } + let text = std::str::from_utf8(&incoming.body) + .map(|t| t.chars().take(200).collect::()) + .unwrap_or_default(); + replies.push(( + address.to_owned(), + format!("twtxt: {reason}"), + format!( + "Your mail could not be applied.\n\nReason: {reason}\n\nQuoted mail:\n{text}\n" + ), + )); + } + + fn send_replies(&self, replies: Vec<(String, String, String)>) -> Result<(), String> { + for (address, subject, body) in replies { + let addr = Address::parse(&address).map_err(|e| format!("reply address: {e}"))?; + client::send( + &self.store, + &self.account, + &client::SendDraft { + address: &addr, + text: body, + subject: Some(&subject), + reply_to: None, + headers: &[], + anonymous: false, + no_pad: true, + }, + self.cfg.timeout, + ) + .map_err(|e: FumiError| format!("reply to {address}: {e}"))?; + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::Config; + use crate::state::State; + use smolboid_core::verify; + + fn feed_state() -> FeedState { + FeedState { + username: "tobias".into(), + address: "tobias@smol.place".into(), + owner_keys: vec![fumi_core::crypto::b32(&[1; 32])], + nick: "tobias".into(), + prune: PruneMode::Off, + warned: false, + day: State::day(0), + posts: 0, + reposts: 0, + } + } + + /// A record renders at roughly 260 bytes: the post line plus a + /// `# smol:mail` line of three base32 values. + fn config(cap: usize) -> Config { + Config { + cap_bytes: cap, + ..Config::default() + } + } + + fn record(ts: &str, text: &str) -> PendingRecord { + PendingRecord { + timestamp: ts.into(), + text: text.into(), + repost: false, + reply_to: None, + mail: MailLine { + sender: [1; 32], + epk: [2; 32], + sig: [3; 64], + }, + } + } + + #[test] + fn posts_publish_and_verify() { + let mut feed = smolboid_core::Feed::new(); + let mut fs = feed_state(); + apply_record(&mut feed, &mut fs, &config(1 << 20), record("2026-10-01T12:47:00Z", "hello"), 0).unwrap(); + assert_eq!(fs.posts, 1); + let report = verify(&feed); + assert_eq!(report.states, vec![smolboid_core::State::Unsigned]); + assert_eq!(feed.entries().len(), 1); + } + + #[test] + fn reposts_publish_as_repost_records() { + let mut feed = smolboid_core::Feed::new(); + let mut fs = feed_state(); + let record = PendingRecord { + timestamp: "2026-10-01T12:47:00Z".into(), + text: "@ #abc1234".into(), + repost: true, + reply_to: None, + mail: MailLine { sender: [1; 32], epk: [2; 32], sig: [3; 64] }, + }; + apply_record(&mut feed, &mut fs, &config(1 << 20), record, 0).unwrap(); + let entry = &feed.entries()[0]; + assert!(matches!(&entry.record, Record::Repost { args, .. } if args == "@ #abc1234")); + assert_eq!(fs.reposts, 1); + assert_eq!(fs.posts, 0); + assert!(feed.render().starts_with("# smol:repost 2026-10-01T12:47:00Z @ #abc1234\n")); + } + + #[test] + fn collisions_and_quotas_reject_without_appending() { + let mut feed = smolboid_core::Feed::new(); + let mut fs = feed_state(); + let cfg = config(1 << 20); + apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:00Z", "one"), 0).unwrap(); + assert!(matches!( + apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:00Z", "two"), 0), + Err(ApplyError::Collision(_)) + )); + assert_eq!(feed.entries().len(), 1); + + let mut fs = feed_state(); + fs.posts = 50; + assert!(matches!( + apply_record(&mut feed, &mut fs, &config(1 << 20), record("2026-10-01T12:48:00Z", "x"), 0), + Err(ApplyError::OverQuota(_)) + )); + // A new UTC day rolls the counters. + let mut fs = feed_state(); + fs.posts = 50; + apply_record(&mut feed, &mut fs, &config(1 << 20), record("2026-10-01T12:48:00Z", "x"), 1_760_000_000).unwrap(); + } + + #[test] + fn hard_limit_rejects_and_fifo_prunes_oldest() { + let mut feed = smolboid_core::Feed::new(); + let mut fs = feed_state(); + let cfg = config(300); + apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:00Z", "first post here"), 0).unwrap(); + assert!(matches!( + apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:48:00Z", "second post"), 0), + Err(ApplyError::FeedFull) + )); + assert_eq!(feed.entries().len(), 1, "off mode removes nothing"); + + fs.prune = PruneMode::Fifo; + apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:48:00Z", "second post"), 0).unwrap(); + let entries = feed.entries(); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].record.timestamp(), "2026-10-01T12:48:00Z"); + // The quota counts published posts; pruning does not refund it. + assert_eq!(fs.posts, 2); + } + + #[test] + fn warning_flag_flips_at_ninety_percent() { + let mut feed = smolboid_core::Feed::new(); + let mut fs = feed_state(); + let cfg = config(1300); + apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:00Z", "x"), 0).unwrap(); + assert!(!fs.warned); + apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:01Z", &"y".repeat(700)), 0).unwrap(); + assert!(fs.warned, "crossing 90% sets the flag once"); + feed.delete("2026-10-01T12:47:01Z"); + apply_record(&mut feed, &mut fs, &cfg, record("2026-10-01T12:47:02Z", "z"), 0).unwrap(); + assert!(!fs.warned, "shrinking below the threshold resets it"); + } + + #[test] + fn commands_touch_feed_and_state() { + let mut feed = smolboid_core::Feed::new(); + let mut fs = feed_state(); + apply_record(&mut feed, &mut fs, &config(1 << 20), record("2026-10-01T12:47:00Z", "gone soon"), 0).unwrap(); + + apply_commands( + &mut feed, + &mut fs, + &[Command::Nick("toby".into()), Command::Prune(PruneMode::Fifo)], + ) + .unwrap(); + assert_eq!(fs.nick, "toby"); + assert_eq!(feed.meta("nick"), Some("toby")); + assert_eq!(fs.prune, PruneMode::Fifo); + + apply_commands(&mut feed, &mut fs, &[Command::Delete("2026-10-01T12:47:00Z".into())]).unwrap(); + assert_eq!(feed.entries().len(), 0); + assert!(apply_commands(&mut feed, &mut fs, &[Command::Delete("2026-10-01T12:47:00Z".into())]).is_err()); + } +} diff --git a/daemon/src/main.rs b/daemon/src/main.rs new file mode 100644 index 0000000..fb5089b --- /dev/null +++ b/daemon/src/main.rs @@ -0,0 +1,301 @@ +//! smolboid: serve twtxt feeds published by smolmail (PLAN.md). +//! +//! Bootstrap with the fumi CLI against the same master key and store before +//! the first `serve`: `fumi --key --db /mail.db trust +//! ` then `fumi ... register @`. + +mod config; +mod dispatch; +mod feedstore; +mod http; +mod mailloop; +mod state; + +use std::path::{Path, PathBuf}; +use std::sync::Mutex; + +use config::Config; +use feedstore::FeedStore; +use fumi_core::account::Account; +use fumi_core::address::Address; +use fumi_core::client::{self, Pushed}; +use fumi_core::crypto::{b32, fingerprint}; +use fumi_core::store::Store; +use mailloop::Daemon; +use state::{FeedState, State}; + +const USAGE: &str = "\ +smolboid — twtxt feeds published by smolmail + + smolboid serve --config run the daemon (default ./smolboid.toml) + smolboid admit --config
verify, accept, and create the feed + smolboid block --config
withdraw the accept token + smolboid keygen write a fresh 32-byte master + smolboid check verify a feed's records +"; + +fn main() { + let args: Vec = std::env::args().skip(1).collect(); + let rest: &[String] = &args; + let command = args.first().map(String::as_str).unwrap_or(""); + let result = match command { + "serve" => serve(config_of(&rest[1..])), + "admit" => admit(config_of(&rest[1..])), + "block" => block(config_of(&rest[1..])), + "keygen" if rest.len() == 2 => keygen(&PathBuf::from(&rest[1])), + "check" if rest.len() == 2 => check(&PathBuf::from(&rest[1])), + _ => { + println!("{USAGE}"); + return; + } + }; + if let Err(message) = result { + eprintln!("smolboid: {message}"); + std::process::exit(1); + } +} + +/// The last non-flag argument of `admit`/`block` is the address; `--config` +/// or the default file names the instance. +fn config_of(rest: &[String]) -> ConfigAndArg { + let mut path = PathBuf::from("smolboid.toml"); + let mut arg = String::new(); + let mut iter = rest.iter(); + while let Some(flag) = iter.next() { + match flag.as_str() { + "--config" => path = PathBuf::from(iter.next().expect("--config needs a file")), + other => arg = other.to_owned(), + } + } + ConfigAndArg { path, arg } +} + +struct ConfigAndArg { + path: PathBuf, + arg: String, +} + +fn load(config: &ConfigAndArg) -> Result<(Config, Store, Account, State, FeedStore), String> { + let cfg = Config::load(&config.path)?; + std::fs::create_dir_all(&cfg.data).map_err(|e| format!("data dir: {e}"))?; + let master = std::fs::read(&cfg.master) + .map_err(|e| format!("master at {}: {e}; run keygen", cfg.master.display()))?; + let master: [u8; 32] = master + .try_into() + .map_err(|v: Vec| format!("master is {} bytes, want 32", v.len()))?; + let store = Store::open(&cfg.data.join("mail.db")).map_err(|e| format!("store: {e}"))?; + let account = Account::new(master, store.rotations().map_err(|e| format!("store: {e}"))?) + .map_err(|e| format!("identity: {e}"))?; + let state = State::load(&cfg.data.join("state.json")).map_err(|e| format!("state: {e}"))?; + let feeds = FeedStore::new(&cfg.data).map_err(|e| format!("feeds: {e}"))?; + Ok((cfg, store, account, state, feeds)) +} + +fn serve(config: ConfigAndArg) -> Result<(), String> { + let (cfg, store, account, state, feeds) = load(&config)?; + let daemon = Daemon { + cfg: cfg.clone(), + store, + account, + state: Mutex::new(state), + feeds: feeds.clone(), + }; + let listener = std::net::TcpListener::bind(&cfg.listen) + .map_err(|e| format!("cannot listen on {}: {e}", cfg.listen))?; + println!( + "smolboid serving {} at http://{} as {}@{}", + cfg.base_url, + listener.local_addr().map_err(|e| e.to_string())?, + cfg.mailbox, + cfg.smolmail_host + ); + std::thread::spawn(move || http::serve(listener, feeds)); + loop { + match daemon.poll() { + Ok(summary) => println!("{} {}", now_stamp(), summary), + Err(e) => eprintln!("{} poll: {e}", now_stamp()), + } + std::thread::sleep(std::time::Duration::from_secs(cfg.poll_secs)); + } +} + +fn admit(config: ConfigAndArg) -> Result<(), String> { + let (cfg, store, account, mut state, feeds) = load(&config)?; + let addr = Address::parse(&config.arg).map_err(|e| format!("address: {e}"))?; + + // Verify the author's key against their own server, then accept them. + let mut session = client::connect(&store, &addr, false, cfg.timeout) + .map_err(|e| format!("connecting to {}: {e}", addr.host))?; + let (key, change) = client::trust_key(&store, session.transport(), &addr) + .map_err(|e| format!("resolving {}: {e}", config.arg))?; + session.close(); + store + .accept(&addr.short(), &key) + .map_err(|e| format!("accepting: {e}"))?; + match client::push_tokens(&store, &account, None, cfg.timeout) { + Ok(Pushed::Held(held)) => println!("server now holds {held} accept token(s)"), + Ok(Pushed::NotRegistered) => println!("not registered; tokens travel with the first fetch"), + Err(e) => return Err(format!("pushing tokens: {e}")), + } + println!( + "{}: {} ({:?})", + config.arg, + b32(&key), + change + ); + + // The username is the feed path; a second owner for one username is a + // collision the operator must see. + let existing = state.feed_by_username(&addr.user); + if let Some(existing) = existing { + if existing.address != addr.short() { + return Err(format!( + "username `{}` already has a feed from {}", + addr.user, existing.address + )); + } + println!("feed already exists; re-accepted"); + } + + let feed_state = FeedState { + username: addr.user.clone(), + address: addr.short(), + owner_keys: vec![b32(&key)], + nick: addr.user.clone(), + prune: Default::default(), + warned: false, + day: State::day(0), + posts: 0, + reposts: 0, + }; + if existing.is_none() { + feeds.create( + &feed_state.username, + &feed_state.nick, + &feed_state.address, + &cfg.base_url, + &account.me().pk(), + )?; + state.feeds.push(feed_state); + } + state + .save(&cfg.data.join("state.json")) + .map_err(|e| format!("state: {e}"))?; + println!( + "feed created: {}/twtxt/{}.txt (host key {})", + cfg.base_url, + addr.user, + b32(&account.me().pk()) + ); + + // The welcome mail carries the Accept token (sec 5.8), which is what + // moves the author's posts into the daemon's main tier; without it their + // mail would stay in requests and never publish. + let sent = client::send( + &store, + &account, + &client::SendDraft { + address: &addr, + text: format!( + "Your smolboid feed is live: {}/twtxt/{}.txt\n\n\ + Post by sending mail to this address with the twt as the body. Manage it with\n\ + Twt-Nick, Twt-Delete, Twt-Repost and Twt-Prune frontmatter fields (PLAN.md).", + cfg.base_url, addr.user + ), + subject: Some("twtxt: your feed is ready"), + reply_to: None, + headers: &[], + anonymous: false, + no_pad: true, + }, + cfg.timeout, + ) + .map_err(|e| format!("welcome mail: {e}"))?; + println!( + "welcome mail sent{}; fetch it before your first post", + if sent.token_used { " with the accept token" } else { "" } + ); + Ok(()) +} + +fn block(config: ConfigAndArg) -> Result<(), String> { + let (cfg, store, account, _state, _feeds) = load(&config)?; + if !store.block(&config.arg).map_err(|e| format!("block: {e}"))? { + return Err(format!("{} was never accepted", config.arg)); + } + match client::push_tokens(&store, &account, None, cfg.timeout) { + Ok(Pushed::Held(held)) => println!("server now holds {held} accept token(s)"), + Ok(Pushed::NotRegistered) => println!("not registered; tokens travel with the first fetch"), + Err(e) => return Err(format!("pushing tokens: {e}")), + } + println!( + "blocked {arg}; their mail lands in requests from their next message on", + arg = config.arg + ); + Ok(()) +} + +fn keygen(path: &Path) -> Result<(), String> { + if path.exists() { + return Err(format!("{} exists; refusing to overwrite", path.display())); + } + let mut master = [0u8; 32]; + std::fs::File::open("/dev/urandom") + .and_then(|mut f| { + use std::io::Read; + f.read_exact(&mut master) + }) + .map_err(|e| format!("urandom: {e}"))?; + std::fs::write(path, master).map_err(|e| format!("cannot write {}: {e}", path.display()))?; + set_private(path); + let account = Account::new(master, 0).map_err(|e| format!("identity: {e}"))?; + println!("master: {} (back this up; it is the only secret)", path.display()); + println!("public key: {}", b32(&account.me().pk())); + println!("fingerprint: {}", fingerprint(&account.me().pk())); + Ok(()) +} + +#[cfg(unix)] +fn set_private(path: &Path) { + use std::os::unix::fs::PermissionsExt; + let _ = std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)); +} + +#[cfg(not(unix))] +fn set_private(_path: &Path) {} + +/// Verifies every record of a feed file and reports the three states; a +/// nonzero exit marks forged records, so scripts can gate on it. +fn check(path: &Path) -> Result<(), String> { + let text = std::fs::read_to_string(path) + .map_err(|e| format!("cannot read {}: {e}", path.display()))?; + let feed = smolboid_core::Feed::parse(&text); + let report = smolboid_core::verify(&feed); + if report.legacy { + println!("legacy feed: no smol metadata, nothing to verify"); + return Ok(()); + } + let entries = feed.entries(); + let mut counts = [0usize; 3]; + for (entry, state) in entries.iter().zip(&report.states) { + let index = match state { + smolboid_core::State::Verified => 0, + smolboid_core::State::Unsigned => 1, + smolboid_core::State::Invalid => 2, + }; + counts[index] += 1; + println!("{} {}", entry.record.timestamp(), state.as_str()); + } + println!( + "{} verified, {} unsigned, {} invalid", + counts[0], counts[1], counts[2] + ); + if counts[2] > 0 { + return Err(format!("{} record(s) do not verify", counts[2])); + } + Ok(()) +} + +fn now_stamp() -> String { + smolboid_core::render_timestamp(fumi_core::store::now()) +} diff --git a/daemon/src/state.rs b/daemon/src/state.rs new file mode 100644 index 0000000..8e5177a --- /dev/null +++ b/daemon/src/state.rs @@ -0,0 +1,237 @@ +//! Persistent instance state: one record per feed (ownership, nick, prune +//! mode, daily counters) plus the reply ledgers. Kept as one JSON file with +//! atomic replace, the feed files being the only other on-disk truth. + +use std::io; +use std::path::Path; + +use fumi_core::crypto::{b32, sha256}; +use serde::{Deserialize, Serialize}; + +/// At most this many replies per sender per hour (PLAN.md, Error replies). +pub const REPLIES_PER_HOUR: usize = 10; +/// Rejected-payload keys remembered for reply dedup, a bounded ring. +pub const DEDUPE_MAX: usize = 256; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FeedState { + pub username: String, + /// The owner's smolmail address, `user@host` short form. + pub address: String, + /// Owner identity keys, oldest first; a rotation appends (SPEC.md §7). + pub owner_keys: Vec, + /// Display nick; the feed path is the username and never changes. + pub nick: String, + pub prune: PruneMode, + /// The 90% cap warning has been sent since the feed last shrank. + pub warned: bool, + /// Daily counters, UTC-day keyed: `YYYY-MM-DD`. + pub day: String, + pub posts: u32, + pub reposts: u32, +} + +impl FeedState { + pub fn latest_owner(&self) -> Option<[u8; 32]> { + self.owner_keys + .last() + .and_then(|k| fumi_core::crypto::unb32(k).ok()) + .and_then(|k| k.try_into().ok()) + } + + /// True when this identity key owns the feed, at any rotation index. + pub fn owned_by(&self, key: &[u8; 32]) -> bool { + self.owner_keys + .iter() + .any(|k| fumi_core::crypto::unb32(k).map(|raw| raw == key).unwrap_or(false)) + } +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +pub enum PruneMode { + /// The default: a full feed rejects new records (PLAN.md, Retention). + #[default] + Off, + /// A full feed drops its oldest records until the new one fits. + Fifo, +} + +#[derive(Debug, Default, Serialize, Deserialize)] +pub struct State { + pub feeds: Vec, + #[serde(default)] + replies: Vec, + #[serde(default)] + dedupe: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +struct ReplyLedger { + sender: String, + times: Vec, +} + +impl State { + pub fn load(path: &Path) -> io::Result { + match std::fs::read_to_string(path) { + Ok(text) => serde_json::from_str(&text).map_err(|e| io::Error::other(e)), + Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(State::default()), + Err(e) => Err(e), + } + } + + pub fn save(&self, path: &Path) -> io::Result<()> { + if let Some(dir) = path.parent() { + std::fs::create_dir_all(dir)?; + } + let tmp = path.with_extension("json.tmp"); + std::fs::write(&tmp, serde_json::to_string_pretty(self).unwrap())?; + std::fs::rename(&tmp, path) + } + + pub fn feed_of(&self, key: &[u8; 32]) -> Option { + self.feeds.iter().position(|f| f.owned_by(key)) + } + + pub fn feed_by_username(&self, username: &str) -> Option<&FeedState> { + self.feeds.iter().find(|f| f.username == username) + } + + /// The UTC day of a timestamp, counter key. + pub fn day(now: i64) -> String { + smolboid_core::render_timestamp(now)[..10].to_owned() + } + + /// Rolls the daily counters when the day changed. + pub fn roll_counters(feed: &mut FeedState, now: i64) { + let today = State::day(now); + if feed.day != today { + feed.day = today; + feed.posts = 0; + feed.reposts = 0; + } + } +} + +/// One rejected mail, identified by signer, payload time and body: an +/// identical resubmission is a new envelope with the same payload, so it is +/// matched here rather than by message id. +pub fn payload_key(sender: &[u8; 32], time: i64, body: &[u8]) -> String { + b32(&sha256(&[sender, &time.to_be_bytes(), body])) +} + +/// Whether an error reply may still be sent: within the per-sender hourly +/// budget and never for a payload already answered. Records the decision. +pub fn reply_allowed(state: &mut State, sender: &[u8; 32], payload: &str, now: i64) -> bool { + if state.dedupe.iter().any(|seen| seen == payload) { + return false; + } + let sender_b32 = b32(sender); + if !state.replies.iter().any(|l| l.sender == sender_b32) { + state.replies.push(ReplyLedger { + sender: sender_b32.clone(), + times: Vec::new(), + }); + } + let ledger = state + .replies + .iter_mut() + .find(|l| l.sender == sender_b32) + .expect("just pushed"); + ledger.times.retain(|t| now - t < 3600); + if ledger.times.len() >= REPLIES_PER_HOUR { + return false; + } + ledger.times.push(now); + state.dedupe.push(payload.to_owned()); + let excess = state.dedupe.len().saturating_sub(DEDUPE_MAX); + state.dedupe.drain(..excess); + true +} + +#[cfg(test)] +mod tests { + use super::*; + + fn feed(username: &str, key: u8) -> FeedState { + FeedState { + username: username.into(), + address: format!("{username}@smol.place"), + owner_keys: vec![b32(&[key; 32])], + nick: username.into(), + prune: PruneMode::default(), + warned: false, + day: State::day(0), + posts: 0, + reposts: 0, + } + } + + #[test] + fn finds_feeds_by_any_rotation_era_key() { + let mut fs = feed("tobias", 1); + let rotated = b32(&[2u8; 32]); + assert!(!fs.owned_by(&[2; 32])); + fs.owner_keys.push(rotated); + let mut state = State::default(); + state.feeds.push(fs); + assert_eq!(state.feed_of(&[1; 32]), Some(0)); + assert_eq!(state.feed_of(&[2; 32]), Some(0)); + assert_eq!(state.feed_of(&[3; 32]), None); + assert_eq!(state.feed_by_username("tobias").unwrap().address, "tobias@smol.place"); + } + + #[test] + fn counters_roll_over_at_utc_midnight() { + let mut fs = feed("tobias", 1); + fs.day = "1970-01-01".into(); + fs.posts = 9; + State::roll_counters(&mut fs, 1_760_000_000); + assert_eq!(fs.posts, 0); + assert_eq!(fs.day, State::day(1_760_000_000)); + fs.posts = 1; + State::roll_counters(&mut fs, 1_760_000_000); + assert_eq!(fs.posts, 1, "same day keeps counters"); + } + + #[test] + fn reply_budget_and_dedupe() { + let mut state = State::default(); + let sender = [7u8; 32]; + let payload = payload_key(&sender, 0, b"body"); + for i in 0..REPLIES_PER_HOUR { + let p = payload_key(&sender, i as i64, b"body"); + assert!(reply_allowed(&mut state, &sender, &p, 1000)); + } + // Budget exhausted for this hour. + assert!(!reply_allowed( + &mut state, + &sender, + &payload_key(&sender, 99, b"other"), + 1000 + )); + // An identical payload never draws a second reply, even later. + assert!(!reply_allowed(&mut state, &sender, &payload, 5000)); + // A new hour opens the budget again, for a new payload. + assert!(reply_allowed( + &mut state, + &sender, + &payload_key(&sender, 100, b"fresh"), + 1000 + 3601 + )); + } + + #[test] + fn state_round_trips_through_json() { + let mut state = State::default(); + state.feeds.push(feed("tobias", 1)); + let path = std::env::temp_dir().join("smolboid-state-test.json"); + state.save(&path).unwrap(); + let back = State::load(&path).unwrap(); + assert_eq!(back.feeds[0].username, "tobias"); + assert!(State::load(&std::env::temp_dir().join("smolboid-absent.json")) + .unwrap() + .feeds + .is_empty()); + } +}