diff --git a/Cargo.toml b/Cargo.toml index 5403e06..eaaee71 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -62,6 +62,10 @@ russh-sftp = "3" # speaks rand_core 0.10, one major version past the `rand` used everywhere else. rand10 = { package = "rand", version = "0.10" } +[dev-dependencies] +# A paused clock, so tests of timed loops do not wait in real time. +tokio = { version = "1", features = ["test-util"] } + [features] default = [] # Keep the Proton session in the OS keyring instead of a 0600 file. Needs a diff --git a/README.md b/README.md index 3206cac..84a5398 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,11 @@ spelling still accepted behind it, so a server configured before the rename keeps running untouched. `WORKSPACE_SECRET` signs the session cookie; generate one with -`openssl rand -base64 32`. Changing it logs every device out. The database and +`openssl rand -base64 32`. Changing it logs every device out, and so does +changing `WORKSPACE_PASSWORD` or `:logout!`, which ends every session but +needs no restart. A setting that is set but cannot be read — a number out of +range, a switch that is neither on nor off — stops the start with the reason, +rather than being taken for its default. The database and uploads live under `WORKSPACE_DATA_DIR` (`./data` by default) — that directory is the entire backup, and it is exactly what the Proton Drive mirror copies. @@ -445,11 +449,14 @@ both halves. ## API -Everything except the two share routes requires the session cookie. +Everything except the login, the health check and the two share routes +requires the session cookie. Media answers `Range` requests, so a video seeks, +sealed or not. | method | path | purpose | | --- | --- | --- | -| `POST` | `/api/login`, `/api/logout` | session | +| `GET` | `/api/health` | public: up, and the database answers — for a supervisor | +| `POST` | `/api/login`, `/api/logout` | session; `/api/logout?everywhere=true` ends every one | | `GET` | `/api/me` | current user | | `GET`/`POST` | `/api/spaces` | the spaces / create one | | `GET`/`POST` | `/api/nodes` | children (`?parent=`) or search (`?q=`, `?space=`) / create | @@ -494,7 +501,17 @@ so a proxy must pass the `Host` header through unchanged (traefik and nginx's `proxy_set_header Host $host` both do). Only the upload routes accept a body as large as `WORKSPACE_MAX_UPLOAD_MB`; everything else stops at 16 MiB, and the login at 16 KiB, before anyone has proved who they are. `SIGTERM` gets in-flight -requests eight seconds to finish, inside the ten `docker stop` waits. +requests eight seconds to finish, inside the ten `docker stop` waits. Any +other request that has not been answered after 30 seconds is given up on with +a `503`, except the ones that move files around: uploads, imports, the export, +re-sealing and a backup run. + +Every request is logged at `info`, path and status and time, never a query +string, and with a share link's token and key left out. A background loop that +panics — the mirror, the mail, the sweep, the SFTP server — is logged and +started again rather than lost until the next restart. If +`WORKSPACE_SESSION_KEY_FILE` cannot be read the server starts anyway with +backups off, and `:backup` says why. ### Docker diff --git a/compose.yml b/compose.yml index b54badf..fac9043 100644 --- a/compose.yml +++ b/compose.yml @@ -28,8 +28,9 @@ services: # rather than relying on the reverse proxy — see compose.override.yml.example. # - "127.0.0.1:2222:2222" healthcheck: - # No extra tooling in the image, so probe the port with bash directly. - test: ["CMD", "bash", "-c", "exec 3<>/dev/tcp/127.0.0.1/4242"] + # The image has bash and nothing like curl: a request by hand, and a + # 200 from /api/health, which also asks the database. + test: ["CMD", "bash", "-c", "exec 3<>/dev/tcp/127.0.0.1/4242 && printf 'GET /api/health HTTP/1.0\r\n\r\n' >&3 && head -1 <&3 | grep -q ' 200 '"] interval: 30s timeout: 3s start_period: 5s diff --git a/migrations/0014_session_epoch.sql b/migrations/0014_session_epoch.sql new file mode 100644 index 0000000..0c0a9f3 --- /dev/null +++ b/migrations/0014_session_epoch.sql @@ -0,0 +1,10 @@ +-- Sessions are signed cookies, not rows: nothing on the server lists them, so +-- nothing can revoke one. Signing a number into every cookie that has to match +-- this one gives a single lever instead — raise it, and every cookie issued +-- before is refused. It is raised by `:logout!` and whenever the login +-- password changes, which is recognised by a keyed hash of it kept here. +CREATE TABLE auth ( + id INTEGER PRIMARY KEY CHECK (id = 1), + epoch INTEGER NOT NULL, + password_hash TEXT NOT NULL +); diff --git a/src/auth.rs b/src/auth.rs index 5dede65..f019fe5 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -28,14 +28,16 @@ fn sign(secret: &[u8], payload: &str) -> String { /// A session token is `base64(expiry_unix_seconds).base64(hmac)`. There is no /// server-side session store: rotating `WORKSPACE_SECRET` invalidates every session. -pub fn issue_token(config: &Config) -> String { +/// A cookie is good until it expires and for as long as the epoch it was +/// issued under is the current one. +pub fn issue_token(config: &Config, epoch: i64) -> String { let expires = crate::now() + config.session_days * 86_400; - let payload = B64.encode(expires.to_string()); + let payload = B64.encode(format!("{expires}:{epoch}")); let signature = sign(&config.secret, &payload); format!("{payload}.{signature}") } -pub fn verify_token(config: &Config, token: &str) -> bool { +pub fn verify_token(config: &Config, token: &str, epoch: i64) -> bool { let Some((payload, signature)) = token.split_once('.') else { return false; }; @@ -46,10 +48,61 @@ pub fn verify_token(config: &Config, token: &str) -> bool { let Ok(decoded) = B64.decode(payload) else { return false; }; - let Ok(expires) = String::from_utf8_lossy(&decoded).parse::() else { + let decoded = String::from_utf8_lossy(&decoded); + // A cookie from before epochs existed carries only the expiry, and + // belongs to the first epoch. + let (expires, issued_in) = decoded.split_once(':').unwrap_or((&decoded, "0")); + let (Ok(expires), Ok(issued_in)) = (expires.parse::(), issued_in.parse::()) else { return false; }; - expires > crate::now() + expires > crate::now() && issued_in == epoch +} + +/// The session epoch: what every cookie must have been signed under. +pub struct Epoch; + +impl Epoch { + /// A keyed hash of the login password, so a change to it can be told + /// from a restart without the password being written down anywhere. + fn password_hash(config: &Config) -> String { + sign(&config.secret, &format!("password:{}", config.password)) + } + + /// The epoch to start with: the stored one, raised when the password + /// is not the one it was stored with. + pub async fn load(db: &sqlx::SqlitePool, config: &Config) -> anyhow::Result { + let hash = Self::password_hash(config); + let stored: Option<(i64, String)> = + sqlx::query_as("SELECT epoch, password_hash FROM auth WHERE id = 1") + .fetch_optional(db) + .await?; + let epoch = match stored { + None => 0, + Some((epoch, previous)) if previous == hash => return Ok(epoch), + Some((epoch, _)) => { + tracing::info!("the login password changed: every device is signed out"); + epoch + 1 + } + }; + sqlx::query( + "INSERT INTO auth (id, epoch, password_hash) VALUES (1, ?1, ?2) + ON CONFLICT (id) DO UPDATE SET epoch = excluded.epoch, password_hash = excluded.password_hash", + ) + .bind(epoch) + .bind(&hash) + .execute(db) + .await?; + Ok(epoch) + } + + /// Sign every device out: the epoch moves on, and no cookie has it yet. + pub async fn advance(state: &AppState) -> anyhow::Result { + let epoch: i64 = sqlx::query_scalar("UPDATE auth SET epoch = epoch + 1 WHERE id = 1 RETURNING epoch") + .fetch_one(&state.db) + .await?; + state.epoch.store(epoch, std::sync::atomic::Ordering::SeqCst); + Ok(epoch) + } } pub fn credentials_match(config: &Config, username: &str, password: &str) -> bool { @@ -94,7 +147,8 @@ impl FromRequestParts for Session { state: &AppState, ) -> Result { let token = cookie_value(parts, COOKIE_NAME).ok_or(AppError::Unauthorized)?; - if verify_token(&state.config, token) { + let epoch = state.epoch.load(std::sync::atomic::Ordering::SeqCst); + if verify_token(&state.config, token, epoch) { Ok(Session) } else { Err(AppError::Unauthorized) @@ -199,45 +253,45 @@ mod tests { #[test] fn a_fresh_token_verifies() { let config = config(30); - assert!(verify_token(&config, &issue_token(&config))); + assert!(verify_token(&config, &issue_token(&config, 0), 0)); } #[test] fn an_expired_token_does_not() { let config = config(-1); - assert!(!verify_token(&config, &issue_token(&config))); + assert!(!verify_token(&config, &issue_token(&config, 0), 0)); } #[test] fn rotating_the_secret_invalidates_every_session() { - let token = issue_token(&config(30)); + let token = issue_token(&config(30), 0); let mut rotated = config(30); rotated.secret = b"a-different-secret-entirely".to_vec(); - assert!(!verify_token(&rotated, &token)); + assert!(!verify_token(&rotated, &token, 0)); } #[test] fn a_tampered_token_does_not_verify() { let config = config(30); - let token = issue_token(&config); + let token = issue_token(&config, 0); let (payload, signature) = token.split_once('.').unwrap(); // A far-future expiry the holder signed themselves. let forged = base64::Engine::encode(&super::B64, "99999999999"); - assert!(!verify_token(&config, &format!("{forged}.{signature}"))); + assert!(!verify_token(&config, &format!("{forged}.{signature}"), 0)); // The real expiry with the last byte of the signature flipped. let mut bytes = signature.as_bytes().to_vec(); bytes[0] = if bytes[0] == b'A' { b'B' } else { b'A' }; let flipped = String::from_utf8(bytes).unwrap(); - assert!(!verify_token(&config, &format!("{payload}.{flipped}"))); + assert!(!verify_token(&config, &format!("{payload}.{flipped}"), 0)); } #[test] fn malformed_tokens_are_rejected_rather_than_panicking() { let config = config(30); for token in ["", ".", "no-dot", "!!!.!!!", "a.b.c", &".".repeat(100)] { - assert!(!verify_token(&config, token), "{token:?}"); + assert!(!verify_token(&config, token, 0), "{token:?}"); } } @@ -273,4 +327,19 @@ mod tests { // curl, the CLI, a script with its own credentials. assert!(is_same_origin(&headers(&[host]))); } + + #[test] + fn a_cookie_from_an_earlier_epoch_is_refused() { + let config = config(30); + let token = issue_token(&config, 3); + assert!(verify_token(&config, &token, 3)); + assert!(!verify_token(&config, &token, 4)); + + // Issued before epochs: only the expiry, and good in the first one. + let expires = crate::now() + 3600; + let payload = base64::Engine::encode(&super::B64, expires.to_string()); + let legacy = format!("{payload}.{}", super::sign(&config.secret, &payload)); + assert!(verify_token(&config, &legacy, 0)); + assert!(!verify_token(&config, &legacy, 1)); + } } diff --git a/src/backup/mod.rs b/src/backup/mod.rs index 2aa9fb7..1c0809f 100644 --- a/src/backup/mod.rs +++ b/src/backup/mod.rs @@ -62,16 +62,34 @@ pub struct Backup { /// Held for the duration of a run, so a manual backup and the timer cannot /// upload the same snapshot twice at once. running: tokio::sync::Mutex<()>, + /// Why backups cannot run at all — a session key file that is configured + /// but unreadable — when that is so. The server still starts; the mirror + /// says why it is off, every time it is asked. + disabled: Option, } impl Backup { pub fn new(db: SqlitePool, config: Arc) -> Result> { let store = SessionStore::open(&config)?; - let configured = matches!(store.load(), Ok(Some(_))); - Ok(Arc::new(Self { + Ok(Self::with_store(db, config, store, None)) + } + + /// A mirror that is off for `reason`, for a server that should start + /// anyway: backups are worth a loud error, not an outage of the workspace. + pub fn unavailable(db: SqlitePool, config: Arc, reason: String) -> Arc { + let store = SessionStore::new(config.proton_session_path()); + let backup = Self::with_store(db, config, store, Some(reason.clone())); + backup.set(|s| s.last_error = Some(reason)); + backup + } + + fn with_store(db: SqlitePool, config: Arc, store: SessionStore, disabled: Option) -> Arc { + let configured = disabled.is_none() && matches!(store.load(), Ok(Some(_))); + Arc::new(Self { db, config, store, + disabled, wake: Notify::new(), // The workspace may have been written to while this process was down, // and a restart is the cheapest moment to find out. @@ -81,11 +99,11 @@ impl Backup { ..Status::default() }), running: tokio::sync::Mutex::new(()), - })) + }) } pub fn status(&self) -> Status { - let mut status = self.status.lock().expect("backup status mutex poisoned").clone(); + let mut status = self.status.lock().unwrap_or_else(|e| e.into_inner()).clone(); status.pending = self.dirty.load(Ordering::Relaxed); status } @@ -100,7 +118,20 @@ impl Backup { /// session has been stored — asking an unconfigured backup to run is a /// no-op, not a failure. pub async fn run_now(&self) -> Result> { + if let Some(reason) = &self.disabled { + anyhow::bail!("backups are off: {reason}"); + } let _guard = self.running.lock().await; + // The same, across processes: `backup-now` beside a running server + // would otherwise share the snapshot file and the bookkeeping with it. + let _lock = match self.lock_data_dir() { + Ok(lock) => lock, + Err(e) => { + let message = format!("{e:#}"); + self.set(move |s| s.last_error = Some(message)); + return Err(e); + } + }; self.dirty.store(false, Ordering::Relaxed); self.set(|s| { s.running = true; @@ -145,6 +176,23 @@ impl Backup { outcome } + fn lock_data_dir(&self) -> Result { + let path = self.config.data_dir.join(".backup.lock"); + let file = std::fs::OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(&path) + .with_context(|| format!("could not open {}", path.display()))?; + match file.try_lock() { + Ok(()) => Ok(file), + Err(std::fs::TryLockError::WouldBlock) => { + anyhow::bail!("another backup of this data directory is running — `backup-now` and the server take turns") + } + Err(std::fs::TryLockError::Error(e)) => Err(e).context("could not lock the data directory for a backup"), + } + } + async fn mirror(&self) -> Result> { let Some(client) = proton::resume(&self.store).await? else { return Ok(None); @@ -285,50 +333,63 @@ impl Backup { } fn set(&self, edit: impl FnOnce(&mut Status)) { - let mut status = self.status.lock().expect("backup status mutex poisoned"); + let mut status = self.status.lock().unwrap_or_else(|e| e.into_inner()); edit(&mut status); } } -/// Run the mirror in the background for as long as the server lives. +/// Run the mirror in the background for as long as the server lives. With +/// no Proton session yet it waits for one: `proton-login` is a separate +/// process, and the server picks its session up without a restart. pub fn spawn(backup: Arc) { if backup.store.load().ok().flatten().is_none() { tracing::info!( "Proton Drive backups are not configured — run `narl-workspace proton-login` to enable them" ); + } + crate::tasks::supervise("the backup", move || mirror_forever(Arc::clone(&backup))); +} + +async fn mirror_forever(backup: Arc) { + if backup.disabled.is_some() { return; } + let interval = backup.config.backup.interval; + let debounce = backup.config.backup.debounce; - tokio::spawn(async move { - let interval = backup.config.backup.interval; - let debounce = backup.config.backup.debounce; - - loop { - // Nothing to do: wait for a change, or for the backstop timer. - while !backup.dirty.load(Ordering::Relaxed) { - match interval { - Some(interval) => { - tokio::select! { - _ = backup.wake.notified() => {} - _ = tokio::time::sleep(interval) => break, - } - } - None => backup.wake.notified().await, - } + loop { + if backup.store.load().ok().flatten().is_none() { + tokio::select! { + _ = backup.wake.notified() => {} + _ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {} } - - // Let the workspace fall quiet. Every further change restarts the wait, - // so a writing session is mirrored once, when it is over. - loop { - tokio::select! { - _ = backup.wake.notified() => continue, - _ = tokio::time::sleep(debounce) => break, - } - } - - let _ = backup.run_now().await; + continue; } - }); + + // Nothing to do: wait for a change, or for the backstop timer. + while !backup.dirty.load(Ordering::Relaxed) { + match interval { + Some(interval) => { + tokio::select! { + _ = backup.wake.notified() => {} + _ = tokio::time::sleep(interval) => break, + } + } + None => backup.wake.notified().await, + } + } + + // Let the workspace fall quiet. Every further change restarts the wait, + // so a writing session is mirrored once, when it is over. + loop { + tokio::select! { + _ = backup.wake.notified() => continue, + _ = tokio::time::sleep(debounce) => break, + } + } + + let _ = backup.run_now().await; + } } /// One-shot mirror for `narl-workspace backup-now`, without starting a server. diff --git a/src/config.rs b/src/config.rs index e1674b2..49d105d 100644 --- a/src/config.rs +++ b/src/config.rs @@ -6,7 +6,7 @@ use chrono_tz::Tz; /// Runtime configuration. Everything comes from the environment (`.env` is /// loaded on startup), because this is a deliberately single-user application. -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct Config { pub username: String, pub password: String, @@ -23,6 +23,61 @@ pub struct Config { pub sftp: SftpConfig, } +/// Written by hand so that a `{:?}` anywhere — a log line, a panic — never +/// prints the password, the cookie secret or the relay's credentials. +impl std::fmt::Debug for Config { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Config") + .field("username", &self.username) + .field("password", &"") + .field("secret", &"") + .field("bind", &self.bind) + .field("data_dir", &self.data_dir) + .field("session_days", &self.session_days) + .field("max_upload_bytes", &self.max_upload_bytes) + .field("secure_cookie", &self.secure_cookie) + .field("vault_idle", &self.vault_idle) + .field("backup", &self.backup) + .field("mail", &self.mail) + .field("sftp", &self.sftp) + .finish() + } +} + +/// A whole number from the environment, `default` when it is not set, and an +/// error at startup — not an overflow later — when it is outside `allowed`. +fn number(key: &str, default: u64, allowed: std::ops::RangeInclusive) -> Result { + let Some(raw) = var(key) else { + return Ok(default); + }; + let value: u64 = raw + .trim() + .parse() + .with_context(|| format!("WORKSPACE_{key} must be a whole number, not {raw:?}"))?; + if !allowed.contains(&value) { + anyhow::bail!( + "WORKSPACE_{key} must be between {} and {}, not {value}", + allowed.start(), + allowed.end() + ); + } + Ok(value) +} + +/// A switch. Anything that is not clearly on or clearly off is an error: +/// `WORKSPACE_SECURE_COOKIE=yes` quietly meaning "no" is how a cookie ends up +/// going out over plain HTTP. +fn flag(key: &str) -> Result { + let Some(raw) = var(key) else { + return Ok(false); + }; + match raw.trim().to_ascii_lowercase().as_str() { + "1" | "true" | "yes" | "on" => Ok(true), + "0" | "false" | "no" | "off" => Ok(false), + _ => anyhow::bail!("WORKSPACE_{key} must be 1 or 0 (true/false, yes/no, on/off), not {raw:?}"), + } +} + /// The SFTP mount. Dormant until `WORKSPACE_SFTP_BIND` is set — same shape as /// the mail relay, because a workspace with no key registered yet should not /// refuse to start over a feature nobody has turned on. @@ -51,7 +106,7 @@ pub struct TimeOfDay { /// Notification mail. Dormant until `WORKSPACE_SMTP_URL` is set — the same /// dormant-until-configured shape the Proton mirror has, because a workspace /// that refuses to start over an unconfigured mail relay would be a bad trade. -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct MailConfig { /// `smtp://host:587` or `smtps://user:pass@host:465`. `None` turns every /// notification off, including the queue that would otherwise pile up. @@ -71,6 +126,22 @@ pub struct MailConfig { pub kinds: Vec, } +impl std::fmt::Debug for MailConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("MailConfig") + // The URL carries the relay's user and password. + .field("url", &self.url.as_ref().map(|_| "")) + .field("from", &self.from) + .field("to", &self.to) + .field("timezone", &self.timezone) + .field("reminder_at", &self.reminder_at) + .field("card_due_at", &self.card_due_at) + .field("digest_at", &self.digest_at) + .field("kinds", &self.kinds) + .finish() + } +} + impl MailConfig { /// Whether mail is configured at all. pub fn enabled(&self) -> bool { @@ -190,32 +261,16 @@ impl Config { let data_dir = PathBuf::from(var("DATA_DIR").unwrap_or_else(|| "./data".into())); - let session_days = var("SESSION_DAYS") - .map(|v| v.parse()) - .transpose() - .context("WORKSPACE_SESSION_DAYS must be a number")? - .unwrap_or(30); - - let max_upload_bytes = var("MAX_UPLOAD_MB") - .map(|v| v.parse::()) - .transpose() - .context("WORKSPACE_MAX_UPLOAD_MB must be a number")? - .unwrap_or(1024) - * 1024 - * 1024; - - let vault_idle = std::time::Duration::from_secs( - var("VAULT_IDLE_MIN") - .map(|v| v.parse::()) - .transpose() - .context("WORKSPACE_VAULT_IDLE_MIN must be a number")? - .unwrap_or(15) - * 60, - ); - - let secure_cookie = var("SECURE_COOKIE") - .map(|v| v == "1" || v.eq_ignore_ascii_case("true")) - .unwrap_or(false); + // Ten years at most: a session is a cookie, and the expiry is + // arithmetic on it that must not overflow. + let session_days = number("SESSION_DAYS", 30, 1..=3650)? as i64; + // Up to a terabyte, in bytes. + let max_upload_bytes = usize::try_from(number("MAX_UPLOAD_MB", 1024, 1..=1024 * 1024)? * 1024 * 1024) + .context("WORKSPACE_MAX_UPLOAD_MB is more than this machine can address")?; + // A vault that locks the moment it opens could not be used; one that + // stays open for more than a day is not much of a vault. + let vault_idle = Duration::from_secs(number("VAULT_IDLE_MIN", 15, 1..=24 * 60)? * 60); + let secure_cookie = flag("SECURE_COOKIE")?; Ok(Self { username, @@ -311,24 +366,11 @@ impl BackupConfig { // second device and a fresh upload of everything in it. let device_name = var("PROTON_DEVICE").unwrap_or_else(|| "narl-diary".to_string()); - let minutes = var("BACKUP_INTERVAL_MIN") - .map(|v| v.parse::()) - .transpose() - .context("WORKSPACE_BACKUP_INTERVAL_MIN must be a number of minutes")? - .unwrap_or(60); + // 0 turns the backstop off; a week is the longest it waits. + let minutes = number("BACKUP_INTERVAL_MIN", 60, 0..=7 * 24 * 60)?; let interval = (minutes > 0).then(|| Duration::from_secs(minutes * 60)); - - let debounce = Duration::from_secs( - var("BACKUP_DEBOUNCE_SEC") - .map(|v| v.parse::()) - .transpose() - .context("WORKSPACE_BACKUP_DEBOUNCE_SEC must be a number of seconds")? - .unwrap_or(300), - ); - - let prune = var("BACKUP_PRUNE") - .map(|v| v == "1" || v.eq_ignore_ascii_case("true")) - .unwrap_or(false); + let debounce = Duration::from_secs(number("BACKUP_DEBOUNCE_SEC", 300, 0..=24 * 60 * 60)?); + let prune = flag("BACKUP_PRUNE")?; Ok(Self { device_name, @@ -342,9 +384,34 @@ impl BackupConfig { #[cfg(test)] mod tests { - use super::{parse_digest, parse_time, TimeOfDay}; + use super::{flag, number, parse_digest, parse_time, TimeOfDay}; use chrono::Weekday; + /// Each case under a name of its own: tests run in parallel and share + /// one environment. + fn with(key: &str, value: &str) -> String { + std::env::set_var(format!("WORKSPACE_{key}"), value); + key.to_string() + } + + #[test] + fn numbers_out_of_bounds_stop_the_start() { + assert_eq!(number("TEST_UNSET_NUMBER", 7, 1..=10).unwrap(), 7); + assert_eq!(number(&with("TEST_NUMBER_OK", " 9 "), 7, 1..=10).unwrap(), 9); + assert!(number(&with("TEST_NUMBER_ZERO", "0"), 7, 1..=10).is_err()); + assert!(number(&with("TEST_NUMBER_HUGE", "99999999999999999999"), 7, 1..=10).is_err()); + assert!(number(&with("TEST_NUMBER_NEGATIVE", "-1"), 7, 1..=10).is_err()); + } + + #[test] + fn a_switch_is_on_off_or_an_error() { + assert!(!flag("TEST_UNSET_FLAG").unwrap()); + assert!(flag(&with("TEST_FLAG_YES", "yes")).unwrap()); + assert!(flag(&with("TEST_FLAG_ONE", "1")).unwrap()); + assert!(!flag(&with("TEST_FLAG_OFF", "off")).unwrap()); + assert!(flag(&with("TEST_FLAG_TYPO", "ture")).is_err()); + } + #[test] fn a_time_of_day_is_read_as_written() { assert_eq!( diff --git a/src/main.rs b/src/main.rs index 3021da2..913cda6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -11,6 +11,7 @@ mod sftp; mod state; mod static_files; mod sweep; +mod tasks; mod throttle; mod vault; @@ -23,7 +24,9 @@ use sqlx::sqlite::{ }; use axum::http::{header, HeaderValue}; use tower_http::{ - compression::CompressionLayer, set_header::SetResponseHeaderLayer, trace::TraceLayer, + compression::CompressionLayer, + set_header::SetResponseHeaderLayer, + trace::{DefaultOnResponse, TraceLayer}, }; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter}; @@ -106,7 +109,14 @@ async fn main() -> Result<()> { let bind = config.bind; let max_upload = config.max_upload_bytes; let config = Arc::new(config); - let backup = backup::Backup::new(db.clone(), Arc::clone(&config))?; + let backup = match backup::Backup::new(db.clone(), Arc::clone(&config)) { + Ok(backup) => backup, + Err(e) => { + let reason = format!("{e:#}"); + tracing::error!("Proton Drive backups are off: {reason}"); + backup::Backup::unavailable(db.clone(), Arc::clone(&config), reason) + } + }; backup::spawn(Arc::clone(&backup)); let notify = notify::Notifier::new(db.clone(), Arc::clone(&config))?; @@ -114,6 +124,7 @@ async fn main() -> Result<()> { sweep::spawn(db.clone(), config.uploads_dir()); + let epoch = auth::Epoch::load(&db, &config).await?; let vault = vault::Vault::new(db.clone(), config.vault_idle); vault.spawn_idle_lock(); let state = AppState { @@ -124,13 +135,20 @@ async fn main() -> Result<()> { vault_throttle: Arc::default(), backup, notify, + epoch: Arc::new(std::sync::atomic::AtomicI64::new(epoch)), }; if let Some(sftp_bind) = config.sftp.bind { let sftp_state = state.clone(); - tokio::spawn(async move { - if let Err(e) = sftp::serve(sftp_state, sftp_bind).await { - tracing::error!("sftp server stopped: {e:#}"); + tasks::supervise("the sftp server", move || { + let state = sftp_state.clone(); + async move { + // A port that is taken now may be free in a minute; the web + // side carries on either way. + while let Err(e) = sftp::serve(state.clone(), sftp_bind).await { + tracing::error!("sftp server stopped: {e:#}; trying again in a minute"); + tokio::time::sleep(std::time::Duration::from_secs(60)).await; + } } }); } @@ -222,9 +240,21 @@ async fn serve(bind: std::net::SocketAddr, max_upload: usize, state: AppState) - .nest("/api", routes::api_router(max_upload)) .fallback(static_files::handler) .layer(axum::middleware::from_fn(auth::same_origin)) - .layer(CompressionLayer::new()) + .layer(axum::middleware::from_fn(deadline)) + .layer(CompressionLayer::new().compress_when(compressible)) .layer(security_headers()) - .layer(TraceLayer::new_for_http()) + .layer( + TraceLayer::new_for_http() + .make_span_with(|request: &axum::http::Request| { + tracing::info_span!( + "request", + method = %request.method(), + path = %loggable(request.uri().path()), + ) + }) + .on_request(()) + .on_response(DefaultOnResponse::new().level(tracing::Level::INFO)), + ) .with_state(state); let listener = tokio::net::TcpListener::bind(bind) @@ -250,9 +280,90 @@ async fn serve(bind: std::net::SocketAddr, max_upload: usize, state: AppState) - Ok(()) } +/// How long an ordinary request may take to be answered. A request that takes +/// this long is stuck — waiting on a lock it will not get — and is better +/// answered with an error than left hanging. +const REQUEST_DEADLINE: std::time::Duration = std::time::Duration::from_secs(30); + +/// Requests that take as long as the bytes do: uploads, imports, re-sealing, +/// a backup run, the export. Only the time to the first byte of a response is +/// bounded anyway, so a long download of a file is never cut off. +fn is_slow(method: &axum::http::Method, path: &str) -> bool { + use axum::http::Method; + match *method { + Method::POST => { + path == "/api/media" + || path == "/api/import" + || path == "/api/backup" + || path.ends_with("/import") + || path.ends_with("/files") + || path.ends_with("/encrypt") + || path.ends_with("/decrypt") + } + Method::GET => path == "/api/export", + _ => false, + } +} + +async fn deadline(request: axum::extract::Request, next: axum::middleware::Next) -> axum::response::Response { + use axum::response::IntoResponse; + if is_slow(request.method(), request.uri().path()) { + return next.run(request).await; + } + let path = loggable(request.uri().path()).to_string(); + match tokio::time::timeout(REQUEST_DEADLINE, next.run(request)).await { + Ok(response) => response, + Err(_) => { + tracing::error!("{path} took longer than {REQUEST_DEADLINE:?} and was given up on"); + ( + axum::http::StatusCode::SERVICE_UNAVAILABLE, + axum::Json(serde_json::json!({ "error": "the server took too long — try again" })), + ) + .into_response() + } + } +} + /// Below the ten seconds `docker stop` waits before it sends SIGKILL. const SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(8); +/// A request path as it may go into the log: a share link's token and key are +/// the whole of its secret, and the key is kept out of every log on purpose — +/// it travels in the URL fragment so that no server ever sees it, except this +/// one's API, which must not write it down either. Query strings, which carry +/// search terms, are never logged at all. +fn loggable(path: &str) -> &str { + if path.starts_with("/api/share/") { + "/api/share/…" + } else if path.starts_with("/s/") { + "/s/…" + } else { + path + } +} + +/// Text is worth compressing; uploads are not. Media is already compressed, +/// and a compressed body no longer matches the byte ranges a video player +/// asks for, so nothing that answers a range — or could — is touched. +fn compressible( + status: axum::http::StatusCode, + _: axum::http::Version, + headers: &axum::http::HeaderMap, + _: &axum::http::Extensions, +) -> bool { + if status == axum::http::StatusCode::PARTIAL_CONTENT || headers.contains_key(header::ACCEPT_RANGES) { + return false; + } + let kind = headers + .get(header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .unwrap_or_default(); + kind.starts_with("text/") + || kind.starts_with("application/json") + || kind.starts_with("application/javascript") + || kind.starts_with("image/svg+xml") +} + /// Content-Security-Policy for the app shell. `unsafe-inline` is granted to /// styles only, because CodeMirror injects its theme as a