diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d71c7e..49d97f0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ content repos pin the tag they run (see the end of this file). ## Unreleased +- Sessions live in NATS KV (`src/sessions.rs`), not in memory. Every restart used to sign out every desk on the site being deployed; they now survive it, in the JetStream that is already under everything else - no new service and no new store to run. The bucket's `max_age` collects what nobody returns for, and a record's own expiry is what answers a load. +- Fixed: the JetStream integration test asserted a record's projected state was one a desk move had never written to it, so it failed whenever it actually ran. `test.yml` has been red since 0.3.47 while `publish.yml` stayed green, which is how 0.4.0 shipped past it. - A file can come back out. `GET /attachment/{bucket}/{record}/{field}` streams what a `type: file` field stored, reading the object key out of the record rather than taking one from the caller - so a file is only reachable through a record somebody may see, and a key that leaks off a desk is a link to nothing. Before this, uploads were one-way: a desk showed the key as a string and there was no way to open it. - Rate limiting, per caller IP, outside everything else: a burst of 30 refilled one every two seconds (`PORTAL_RATE_BURST`, `PORTAL_RATE_SECONDS`; either 0 turns it off). A form-filling visitor never reaches it; a script does. There was none at all before. - `/upload` asks the policy, like every other way in. It asked nothing: any caller who could reach the host could put 20 MB in the bucket, as often as they liked, and a site whose forms are public is a site whose upload route is public. It now takes the `question` and `alternative` the file is being attached to, refuses a page that declares no `type: file` field or an alternative that is closed, and asks `Submit` on that page - the same question the submission itself will ask - before a byte is written. Latent until an instance had storage configured; the first one now does. diff --git a/Cargo.lock b/Cargo.lock index 00158a2..43f6991 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -132,9 +132,9 @@ checksum = "4288f83726785267c6f2ef073a3d83dc3f9b81464e9f99898240cced85fce35a" [[package]] name = "async-trait" -version = "0.1.91" +version = "0.1.92" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" dependencies = [ "proc-macro2", "quote", @@ -3362,6 +3362,7 @@ dependencies = [ "anyhow", "arc-swap", "async-nats", + "async-trait", "aws-sdk-s3", "axum", "base64 0.22.1", diff --git a/Cargo.toml b/Cargo.toml index 6e425d7..33dff7a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -64,6 +64,7 @@ web-sys = { version = "0.3", features = [ "Window", ], optional = true } tower_governor = { version = "0.8.0", optional = true } +async-trait = { version = "0.1.92", optional = true } [features] default = [] @@ -83,6 +84,7 @@ ssr = [ "dep:axum", "dep:aws-sdk-s3", "dep:tower_governor", + "dep:async-trait", "dep:tokio", "dep:tower", "dep:tower-http", diff --git a/docs/architecture.md b/docs/architecture.md index 5eb16b8..d473b67 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -85,3 +85,8 @@ their graphs with no bespoke UI per bucket. - `iris` — the lint, from its own repo, shipped in every release tarball so a content repo's CI can lint with the exact version its instance runs. + +**Sessions** (`src/sessions.rs`) — in the same JetStream KV as +everything else, so a restart does not sign anybody out. The bucket +carries a `max_age` so NATS collects what nobody comes back for, and +each record's own `expiry_date` is what decides on the way out. diff --git a/docs/operations.md b/docs/operations.md index 66e202e..da9d08c 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -66,7 +66,7 @@ values. | | | |---|---| -| `NATS_URL` | events, projections, content reload, access questions | +| `NATS_URL` | events, projections, sessions, content reload, access questions | | `CONTENT_REPO`, `CONTENT_BRANCH` | the repo that is the site | | `KANIDM_URL`, `OAUTH2_CLIENT_ID`, `OAUTH2_CLIENT_SECRET` | sign-in | | `KANIDM_API_TOKEN` | invites and `grants:`; unset, neither happens | diff --git a/src/lib.rs b/src/lib.rs index 5806c98..aef99d5 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -16,6 +16,9 @@ pub mod announce; #[cfg(feature = "ssr")] pub mod deadlines; +#[cfg(feature = "ssr")] +pub mod sessions; + #[cfg(feature = "ssr")] pub mod http; diff --git a/src/main.rs b/src/main.rs index 92703f1..2aff238 100644 --- a/src/main.rs +++ b/src/main.rs @@ -10,7 +10,7 @@ async fn main() -> anyhow::Result<()> { use portal::upload::{self, Garage}; use std::sync::Arc; use tower_http::services::{ServeDir, ServeFile}; - use tower_sessions::{MemoryStore, SessionManagerLayer}; + use tower_sessions::SessionManagerLayer; dotenvy::dotenv().ok(); tracing_subscriber::fmt() @@ -127,7 +127,10 @@ async fn main() -> anyhow::Result<()> { let cookie_secure = std::env::var("COOKIE_SECURE") .map(|v| v == "true" || v == "1") .unwrap_or(false); - let session_layer = SessionManagerLayer::new(MemoryStore::default()) + // Sessions live in the same JetStream as everything else, so a + // deploy no longer signs out every desk on the site it deploys. + let session_store = portal::sessions::NatsStore::open(&state.jetstream).await?; + let session_layer = SessionManagerLayer::new(session_store) .with_secure(cookie_secure) .with_same_site(tower_sessions::cookie::SameSite::Lax) .with_name("portal_session"); diff --git a/src/sessions.rs b/src/sessions.rs new file mode 100644 index 0000000..86493be --- /dev/null +++ b/src/sessions.rs @@ -0,0 +1,122 @@ +//! Sessions in NATS KV. +//! +//! A session used to live in a `MemoryStore`, which meant every +//! restart signed everybody out: a deploy logged out every desk on +//! every site, and the only reason that was bearable is that the desks +//! were three people. This keeps them in the JetStream KV that is +//! already under everything else here - no new service, no new +//! dependency, and the same bucket a second instance of a site would +//! read if there ever is one. +//! +//! Expiry is said twice on purpose. The bucket carries a `max_age`, so +//! NATS collects abandoned sessions without anyone sweeping; and a +//! record's own `expiry_date` is checked on the way out, because the +//! bucket's age is a coarse fence and the record's date is the truth. +#![cfg(feature = "ssr")] + +use tower_sessions::session::{Id, Record}; +use tower_sessions::session_store::{Error, Result}; + +/// How long an untouched session survives in the bucket. Comfortably +/// longer than any expiry `SessionManagerLayer` will set, so the +/// record's own date is always what decides; this is only here so a +/// session nobody ever comes back for does not sit in NATS forever. +const MAX_AGE: std::time::Duration = std::time::Duration::from_secs(60 * 60 * 24 * 40); + +pub const BUCKET: &str = "portal_sessions"; + +/// A new id is drawn this many times before a collision is reported as +/// a backend error. Two random `i128`s colliding is not a thing that +/// happens; this exists so a bucket misbehaving cannot spin forever. +const ID_ATTEMPTS: usize = 8; + +#[derive(Clone)] +pub struct NatsStore { + store: async_nats::jetstream::kv::Store, +} + +impl std::fmt::Debug for NatsStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("NatsStore").field("bucket", &BUCKET).finish() + } +} + +impl NatsStore { + /// Opens the bucket, creating it on first use the way every other + /// bucket here is created. + pub async fn open(js: &async_nats::jetstream::Context) -> anyhow::Result { + let store = match js.get_key_value(BUCKET).await { + Ok(store) => store, + Err(_) => { + js.create_key_value(async_nats::jetstream::kv::Config { + bucket: BUCKET.to_string(), + max_age: MAX_AGE, + history: 1, + ..Default::default() + }) + .await? + } + }; + Ok(Self { store }) + } + + fn encode(record: &Record) -> Result> { + serde_json::to_vec(record).map_err(|e| Error::Encode(e.to_string())) + } +} + +#[async_trait::async_trait] +impl tower_sessions::SessionStore for NatsStore { + /// `create` draws a fresh id until the bucket accepts one as new. + /// `put` would overwrite a stranger's session on a collision; + /// `create` refuses, which is the whole difference. + async fn create(&self, record: &mut Record) -> Result<()> { + for _ in 0..ID_ATTEMPTS { + let bytes = Self::encode(record)?; + match self.store.create(&record.id.to_string(), bytes.into()).await { + Ok(_) => return Ok(()), + Err(_) => record.id = Id::default(), + } + } + Err(Error::Backend(format!( + "could not find an unused session id in {ID_ATTEMPTS} attempts" + ))) + } + + async fn save(&self, record: &Record) -> Result<()> { + let bytes = Self::encode(record)?; + self.store + .put(&record.id.to_string(), bytes.into()) + .await + .map(|_| ()) + .map_err(|e| Error::Backend(e.to_string())) + } + + async fn load(&self, id: &Id) -> Result> { + let Some(bytes) = self + .store + .get(&id.to_string()) + .await + .map_err(|e| Error::Backend(e.to_string()))? + else { + return Ok(None); + }; + let record: Record = + serde_json::from_slice(&bytes).map_err(|e| Error::Decode(e.to_string()))?; + // An expired record is gone as far as anyone asking is + // concerned, whether or not the bucket has collected it yet. + if record.expiry_date <= tower_sessions::cookie::time::OffsetDateTime::now_utc() { + return Ok(None); + } + Ok(Some(record)) + } + + async fn delete(&self, id: &Id) -> Result<()> { + // Purge, not delete: a signed-out session should leave no + // history in the bucket to read back. + self.store + .purge(&id.to_string()) + .await + .map_err(|e| Error::Backend(e.to_string())) + } +} diff --git a/tests/jetstream.rs b/tests/jetstream.rs index d14efd0..e7204ac 100644 --- a/tests/jetstream.rs +++ b/tests/jetstream.rs @@ -68,6 +68,7 @@ async fn a_submission_is_created_moved_refused_and_read_back_through_the_real_en // through the same store the desks read - and moves nothing: the // record is still where the desk left it. let mut record: Answer = serde_json::from_slice(&store.get("rec-1").await.unwrap().expect("record")).unwrap(); + let state_before_the_note = record.state.clone(); let reply = portal::mail::Incoming { v: portal::mail::CONTRACT, id: "mail-1".into(), @@ -86,10 +87,69 @@ async fn a_submission_is_created_moved_refused_and_read_back_through_the_real_en assert_eq!(read_back.notes.len(), 1); assert_eq!(read_back.notes[0].text, "Thursday suits us."); assert_eq!(read_back.notes[0].from, "a@b.example"); - assert_eq!(read_back.state, "quoted", "a reply is a note, never a decision"); + assert_eq!(read_back.state, state_before_the_note, "a reply is a note, never a decision"); // The same reply read twice adds nothing. assert!(portal::mail::note_for(&reply, &read_back).is_err()); // And the event log is untouched by it. let after = load_events(&js, &bucket, "rec-1").await.expect("events"); assert_eq!(after.len(), 3, "a note writes no event"); } + +/// A session outlives the process that made it. The `MemoryStore` this +/// replaced could not: every deploy signed out every desk, and the +/// only proof that this one does not is a store opened twice against +/// the same JetStream, as two runs of the binary would. +#[tokio::test] +async fn a_session_survives_the_process_that_made_it() { + use tower_sessions::session::{Id, Record}; + use tower_sessions::SessionStore; + + let Ok(url) = std::env::var("PORTAL_TEST_NATS_URL") else { + eprintln!("skipped: PORTAL_TEST_NATS_URL is not set"); + return; + }; + let js = async_nats::jetstream::new(async_nats::connect(&url).await.expect("connect")); + let store = portal::sessions::NatsStore::open(&js).await.expect("open"); + + let mut record = Record { + id: Id::default(), + data: Default::default(), + expiry_date: tower_sessions::cookie::time::OffsetDateTime::now_utc() + + tower_sessions::cookie::time::Duration::hours(1), + }; + record.data.insert("who".into(), json!("ingrid")); + store.create(&mut record).await.expect("create"); + + // A second store over the same JetStream is what the next boot of + // the binary has: nothing carried over in memory. + let next_boot = portal::sessions::NatsStore::open(&js).await.expect("reopen"); + let found = next_boot.load(&record.id).await.expect("load").expect("still there"); + assert_eq!(found.data.get("who"), Some(&json!("ingrid"))); + + // `create` never writes over somebody else's session: asked to + // create one whose id is taken, it takes a new id instead. + let mut collide = Record { id: record.id, data: Default::default(), expiry_date: record.expiry_date }; + next_boot.create(&mut collide).await.expect("create beside it"); + assert_ne!(collide.id, record.id, "an id already in use must not be reused"); + assert_eq!( + next_boot.load(&record.id).await.expect("load").expect("original intact").data.get("who"), + Some(&json!("ingrid")), + "the original session must be untouched" + ); + + // An expired record is gone to anyone asking, whatever the bucket + // still holds. + let stale = Record { + id: Id::default(), + data: Default::default(), + expiry_date: tower_sessions::cookie::time::OffsetDateTime::now_utc() + - tower_sessions::cookie::time::Duration::minutes(1), + }; + next_boot.save(&stale).await.expect("save stale"); + assert!(next_boot.load(&stale.id).await.expect("load stale").is_none()); + + // Signing out leaves nothing to read back. + next_boot.delete(&record.id).await.expect("delete"); + assert!(next_boot.load(&record.id).await.expect("load after delete").is_none()); + next_boot.delete(&collide.id).await.ok(); +}