feat(kanban): reauthorize queued replay

req: sync/019

req: security/004

req: auth/005

req: operations/002

examples/kanban/src/main.rs

@@ -1,5 +1,5 @@
 use axum::extract::{Query, State};
-use axum::http::StatusCode;
+use axum::http::{HeaderMap, StatusCode};
 use axum::response::sse::{Event, KeepAlive, Sse};
 use axum::response::IntoResponse;
 use axum::routing::{get, post};
@@ -31,6 +31,49 @@ struct AppState {
     board: Mutex<BoardState>,
     sync: Mutex<SyncState>,
     sync_store: Option<SyncStore>,
+    sync_sessions: SyncSessionTokens,
+}
+
+#[derive(Default)]
+struct SyncSessionTokens {
+    alice_alpha_editor: Option<String>,
+    bob_alpha_viewer: Option<String>,
+    carol_beta_editor: Option<String>,
+}
+
+impl SyncSessionTokens {
+    fn from_env() -> Self {
+        fn token(name: &str) -> Option<String> {
+            std::env::var(name).ok().filter(|value| !value.is_empty())
+        }
+        Self {
+            alice_alpha_editor: token("HEMX_KANBAN_SESSION_ALICE_ALPHA_EDITOR"),
+            bob_alpha_viewer: token("HEMX_KANBAN_SESSION_BOB_ALPHA_VIEWER"),
+            carol_beta_editor: token("HEMX_KANBAN_SESSION_CAROL_BETA_EDITOR"),
+        }
+    }
+
+    fn is_configured(&self) -> bool {
+        self.alice_alpha_editor.is_some()
+            || self.bob_alpha_viewer.is_some()
+            || self.carol_beta_editor.is_some()
+    }
+
+    fn matches(expected: &Option<String>, candidate: &str) -> bool {
+        let Some(expected) = expected else {
+            return false;
+        };
+        if expected.len() != candidate.len() {
+            return false;
+        }
+        expected
+            .bytes()
+            .zip(candidate.bytes())
+            .fold(0_u8, |difference, (left, right)| {
+                difference | (left ^ right)
+            })
+            == 0
+    }
 }
 
 #[derive(Clone)]
@@ -122,6 +165,8 @@ struct SnapshotCard {
 #[derive(Debug)]
 enum SyncRejection {
     BadRequest(&'static str),
+    Unauthorized(&'static str),
+    Forbidden(&'static str),
     Conflict(&'static str),
     Transient,
     Storage,
@@ -129,13 +174,27 @@ enum SyncRejection {
 
 impl IntoResponse for SyncRejection {
     fn into_response(self) -> axum::response::Response {
-        let (status, error) = match self {
-            Self::BadRequest(error) => (StatusCode::BAD_REQUEST, error),
-            Self::Conflict(error) => (StatusCode::CONFLICT, error),
-            Self::Transient => (StatusCode::SERVICE_UNAVAILABLE, "transient sync failure"),
-            Self::Storage => (StatusCode::INTERNAL_SERVER_ERROR, "sync storage failed"),
+        let (status, kind, error) = match self {
+            Self::BadRequest(error) => (StatusCode::BAD_REQUEST, "invalid-command", error),
+            Self::Unauthorized(error) => (StatusCode::UNAUTHORIZED, "authorization-denial", error),
+            Self::Forbidden(error) => (StatusCode::FORBIDDEN, "authorization-denial", error),
+            Self::Conflict(error) => (StatusCode::CONFLICT, "command-conflict", error),
+            Self::Transient => (
+                StatusCode::SERVICE_UNAVAILABLE,
+                "transport-failure",
+                "transient sync failure",
+            ),
+            Self::Storage => (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "storage-failure",
+                "sync storage failed",
+            ),
         };
-        (status, Json(serde_json::json!({ "error": error }))).into_response()
+        (
+            status,
+            Json(serde_json::json!({ "kind": kind, "error": error })),
+        )
+            .into_response()
     }
 }
 
@@ -361,6 +420,7 @@ async fn main() {
         board: Mutex::new(board),
         sync: Mutex::new(sync),
         sync_store,
+        sync_sessions: SyncSessionTokens::from_env(),
     });
 
     let app = Router::new()
@@ -465,9 +525,91 @@ async fn events(Query(params): Query<BTreeMap<String, String>>) -> impl IntoResp
     sse(batches)
 }
 
-// req: sync/001 req: sync/008 req: sync/012
+#[derive(Clone, Copy)]
+struct CurrentSyncPrincipal {
+    principal: &'static str,
+    tenant: &'static str,
+    can_replay: bool,
+}
+
+fn current_sync_principal(
+    headers: &HeaderMap,
+    sessions: &SyncSessionTokens,
+) -> Result<CurrentSyncPrincipal, SyncRejection> {
+    let session = headers
+        .get(axum::http::header::COOKIE)
+        .and_then(|value| value.to_str().ok())
+        .and_then(|cookies| {
+            cookies.split(';').find_map(|cookie| {
+                cookie
+                    .trim()
+                    .strip_prefix("hemx_kanban_session=")
+                    .map(str::trim)
+            })
+        });
+    let Some(session) = session else {
+        if sessions.is_configured() {
+            return Err(SyncRejection::Unauthorized(
+                "current sync session is required",
+            ));
+        }
+        return Ok(CurrentSyncPrincipal {
+            principal: "demo",
+            tenant: "demo",
+            can_replay: true,
+        });
+    };
+    if SyncSessionTokens::matches(&sessions.alice_alpha_editor, session) {
+        return Ok(CurrentSyncPrincipal {
+            principal: "alice",
+            tenant: "alpha",
+            can_replay: true,
+        });
+    }
+    if SyncSessionTokens::matches(&sessions.bob_alpha_viewer, session) {
+        return Ok(CurrentSyncPrincipal {
+            principal: "bob",
+            tenant: "alpha",
+            can_replay: false,
+        });
+    }
+    if SyncSessionTokens::matches(&sessions.carol_beta_editor, session) {
+        return Ok(CurrentSyncPrincipal {
+            principal: "carol",
+            tenant: "beta",
+            can_replay: true,
+        });
+    }
+    Err(SyncRejection::Unauthorized(
+        "current sync session is invalid",
+    ))
+}
+
+fn authorize_sync_replay(
+    principal: CurrentSyncPrincipal,
+    card_id: u64,
+) -> Result<(), SyncRejection> {
+    if principal.principal == "demo" && principal.tenant == "demo" {
+        return Ok(());
+    }
+    if !principal.can_replay {
+        return Err(SyncRejection::Forbidden(
+            "current principal cannot replay commands",
+        ));
+    }
+    let card_tenant = if card_id == 2 { "beta" } else { "alpha" };
+    if principal.tenant != card_tenant {
+        return Err(SyncRejection::Forbidden(
+            "current tenant cannot access command target",
+        ));
+    }
+    Ok(())
+}
+
+// req: sync/001 req: sync/008 req: sync/012 req: sync/019 req: security/004
 async fn sync_command(
     State(state): State<Arc<AppState>>,
+    headers: HeaderMap,
     Query(params): Query<BTreeMap<String, String>>,
 ) -> Result<Json<SyncAcknowledgement>, SyncRejection> {
     let command_id = CommandId::parse(params.get("command_id"))?;
@@ -477,6 +619,8 @@ async fn sync_command(
         .filter(|value| *value > 0)
         .ok_or(SyncRejection::BadRequest("invalid card_id"))?;
     let canonical_column = CanonicalColumn::parse(params.get("column"))?;
+    let principal = current_sync_principal(&headers, &state.sync_sessions)?;
+    authorize_sync_replay(principal, card_id)?;
 
     let mut sync = state.sync.lock().unwrap();
     if let Some(existing) = sync.acknowledgements.get(&command_id) {

examples/kanban/static/sync.js

@@ -23,11 +23,12 @@ let maxObservedInFlight = 0;
 let stopped = false;
 
 class UploadError extends Error {
-  constructor(status, retryable, reason) {
+  constructor(status, retryable, kind, reason) {
     super(`sync upload failed with ${status}`);
     this.name = "UploadError";
     this.status = status;
     this.retryable = retryable;
+    this.kind = kind;
     this.reason = reason;
   }
 }
@@ -190,8 +191,9 @@ async function upload(command) {
       }
       if (!response.ok) {
         const problem = await response.json().catch(() => ({}));
+        const kind = typeof problem.kind === "string" ? problem.kind : "unclassified-rejection";
         const reason = typeof problem.error === "string" ? problem.error : "unclassified rejection";
-        throw new UploadError(response.status, response.status >= 500, reason);
+        throw new UploadError(response.status, response.status >= 500, kind, reason);
       }
       return response.json();
     } catch (error) {
@@ -282,6 +284,7 @@ async function synchronize(command) {
       root.setAttribute("data-sync-uploaded-total", String(uploadedTotal));
       root.setAttribute("data-sync-pending-count", String((await pendingCommands(database)).length));
       root.setAttribute("data-sync-ack-sequence", String(canonical.serverSequence));
+      root.setAttribute("data-sync-ack-command-id", canonical.commandId);
       root.setAttribute("data-sync-canonical-column", canonical.canonicalColumn);
       setPhase("acknowledged", `Command ${command.id} acknowledged in ${canonical.canonicalColumn}.`);
       root.dispatchEvent(new CustomEvent("kanban:sync-acknowledged", { detail: canonical }));
@@ -324,14 +327,23 @@ async function synchronize(command) {
       setOnline(true);
       clearTimeout(leaseTimer);
       root.setAttribute("data-sync-error", error.message);
-      root.setAttribute("data-sync-error-kind", "permanent-rejection");
       root.setAttribute("data-sync-error-status", String(error.status));
-      root.setAttribute("data-sync-error-reason", error.reason);
-      root.setAttribute("data-sync-rejected-command-id", command.id);
       const remaining = await pendingCommands(database);
-      root.setAttribute("data-sync-pending-count", String(remaining.length));
       setManualRetryAvailable(false);
-      setPhase("rejected", `Command ${command.id} was permanently rejected (${error.status}: ${error.reason}); ${remaining.length} durable command${remaining.length === 1 ? " remains" : "s remain"} queued for review.`);
+      if (error.kind === "authorization-denial") {
+        root.setAttribute("data-sync-error-kind", "authorization-denial");
+        root.setAttribute("data-sync-pending-count", "redacted");
+        root.setAttribute("data-sync-redacted-pending", "true");
+        root.removeAttribute("data-sync-error-reason");
+        root.removeAttribute("data-sync-rejected-command-id");
+        setPhase("authorization-denied", "Current session cannot access local queued work. Sign back into the owning account to continue.");
+      } else {
+        root.setAttribute("data-sync-error-kind", "permanent-rejection");
+        root.setAttribute("data-sync-error-reason", error.reason);
+        root.setAttribute("data-sync-rejected-command-id", command.id);
+        root.setAttribute("data-sync-pending-count", String(remaining.length));
+        setPhase("rejected", `Command ${command.id} was permanently rejected (${error.status}: ${error.reason}); ${remaining.length} durable command${remaining.length === 1 ? " remains" : "s remain"} queued for review.`);
+      }
       await releaseUploaderLease(database);
       root.setAttribute("data-sync-leader", "false");
       return;

examples/kanban/tests/browser_e2e.rs

@@ -372,6 +372,199 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack(
     result.and(quit)
 }
 
+#[tokio::test]
+async fn replay_revalidates_current_principal_and_tenant_without_exposing_local_work(
+) -> WebDriverResult<()> {
+    // test req: sync/019 req: security/004 req: auth/005 req: operations/002
+    let app_port = available_port();
+    let app_addr = format!("127.0.0.1:{app_port}");
+    let mut app_command = Command::new(env!("CARGO_BIN_EXE_hemx-kanban-example"));
+    app_command
+        .env("HEMX_KANBAN_ADDR", &app_addr)
+        .env(
+            "HEMX_KANBAN_SESSION_ALICE_ALPHA_EDITOR",
+            "test-token-alice-alpha-editor",
+        )
+        .env(
+            "HEMX_KANBAN_SESSION_BOB_ALPHA_VIEWER",
+            "test-token-bob-alpha-viewer",
+        )
+        .env(
+            "HEMX_KANBAN_SESSION_CAROL_BETA_EDITOR",
+            "test-token-carol-beta-editor",
+        );
+    let _app = TestProcess::start(app_command, "hemx-kanban", &app_addr, STARTUP_TIMEOUT)
+        .expect("start ready hemx-kanban");
+
+    let webdriver_port = available_port();
+    let webdriver_addr = format!("127.0.0.1:{webdriver_port}");
+    let mut webdriver = Command::new("geckodriver");
+    webdriver.arg("--port").arg(webdriver_port.to_string());
+    let _webdriver = TestProcess::start(webdriver, "geckodriver", &webdriver_addr, STARTUP_TIMEOUT)
+        .expect("start ready geckodriver");
+    let mut caps = DesiredCapabilities::firefox();
+    caps.set_headless()?;
+    let driver = WebDriver::new(&format!("http://{webdriver_addr}"), caps).await?;
+
+    let result = async {
+        driver.goto(&format!("http://{app_addr}/")).await?;
+        let seeded = driver
+            .execute_async(
+                r#"
+                const done = arguments[arguments.length - 1];
+                document.cookie = 'hemx_kanban_session=test-token-carol-beta-editor; Path=/; SameSite=Strict';
+                const open = indexedDB.open('hemx-kanban-v1', 2);
+                open.onupgradeneeded = () => {
+                  const database = open.result;
+                  if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' });
+                  if (!database.objectStoreNames.contains('meta')) database.createObjectStore('meta');
+                };
+                open.onsuccess = () => {
+                  const tx = open.result.transaction('commands', 'readwrite');
+                  tx.objectStore('commands').add({
+                    id: 'auth:1', schemaVersion: 2, actor: 'alice-device', session: 'enqueue-session',
+                    causal: 1, kind: 'reorder_card', cardId: '1', targetColumn: 'done',
+                    eventKind: 'click', key: null, enqueuedPrincipal: 'alice', enqueuedTenant: 'alpha',
+                  });
+                  tx.oncomplete = () => done({ seeded: true });
+                  tx.onabort = () => done({ error: tx.error && tx.error.name });
+                };
+                "#,
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(seeded["seeded"], true, "failed to seed auth queue: {seeded}");
+
+        driver.goto(&format!("http://{app_addr}/sync-demo")).await?;
+        wait_until(
+            &driver,
+            "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'authorization-denied'",
+        )
+        .await?;
+        let cross_tenant = driver
+            .execute(
+                "const root = document.querySelector('[data-kanban-sync]'); const retry = root.querySelector('[data-sync-retry]'); return { phase: root.getAttribute('data-sync-phase'), kind: root.getAttribute('data-sync-error-kind'), errorStatus: root.getAttribute('data-sync-error-status'), pending: root.getAttribute('data-sync-pending-count'), redacted: root.getAttribute('data-sync-redacted-pending'), reason: root.getAttribute('data-sync-error-reason'), rejectedId: root.getAttribute('data-sync-rejected-command-id'), retryDisabled: retry.disabled, leakedId: document.body.textContent.includes('auth:1'), status: root.querySelector('[role=status]').textContent }",
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(cross_tenant["phase"], "authorization-denied");
+        assert_eq!(cross_tenant["kind"], "authorization-denial");
+        assert_eq!(cross_tenant["errorStatus"], "403");
+        assert_eq!(cross_tenant["pending"], "redacted");
+        assert_eq!(cross_tenant["redacted"], "true");
+        assert!(cross_tenant["reason"].is_null());
+        assert!(cross_tenant["rejectedId"].is_null());
+        assert_eq!(cross_tenant["retryDisabled"], true);
+        assert_eq!(cross_tenant["leakedId"], false);
+        assert_eq!(
+            cross_tenant["status"],
+            "Current session cannot access local queued work. Sign back into the owning account to continue."
+        );
+        assert_eq!(command_count(&driver).await?, 1);
+
+        driver
+            .execute(
+                "document.cookie = 'hemx_kanban_session=; Path=/; Max-Age=0; SameSite=Strict'; return true;",
+                Vec::new(),
+            )
+            .await?;
+        driver.refresh().await?;
+        wait_until(
+            &driver,
+            "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'authorization-denied'",
+        )
+        .await?;
+        let signed_out = driver
+            .execute(
+                "const root = document.querySelector('[data-kanban-sync]'); return { kind: root.getAttribute('data-sync-error-kind'), status: root.getAttribute('data-sync-error-status'), pending: root.getAttribute('data-sync-pending-count') }",
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(signed_out["kind"], "authorization-denial");
+        assert_eq!(signed_out["status"], "401");
+        assert_eq!(signed_out["pending"], "redacted");
+        assert_eq!(command_count(&driver).await?, 1);
+
+        let before_authorized = driver
+            .execute_async(
+                r#"
+                const done = arguments[arguments.length - 1];
+                fetch('/').then((response) => response.text()).then((html) => {
+                  const page = new DOMParser().parseFromString(html, 'text/html');
+                  done(page.querySelector('[data-key="1"]').closest('section').querySelector('h2').textContent);
+                }).catch((error) => done(`error:${error}`));
+                "#,
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(before_authorized, "Backlog");
+
+        driver
+            .execute(
+                "document.cookie = 'hemx_kanban_session=test-token-bob-alpha-viewer; Path=/; SameSite=Strict'; return true;",
+                Vec::new(),
+            )
+            .await?;
+        driver.refresh().await?;
+        wait_until(
+            &driver,
+            "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'authorization-denied'",
+        )
+        .await?;
+        let stale_permission = driver
+            .execute(
+                "const root = document.querySelector('[data-kanban-sync]'); return { kind: root.getAttribute('data-sync-error-kind'), status: root.getAttribute('data-sync-error-status'), pending: root.getAttribute('data-sync-pending-count'), rejectedId: root.getAttribute('data-sync-rejected-command-id') }",
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(stale_permission["kind"], "authorization-denial");
+        assert_eq!(stale_permission["status"], "403");
+        assert_eq!(stale_permission["pending"], "redacted");
+        assert!(stale_permission["rejectedId"].is_null());
+        assert_eq!(command_count(&driver).await?, 1);
+
+        driver
+            .execute(
+                "document.cookie = 'hemx_kanban_session=test-token-alice-alpha-editor; Path=/; SameSite=Strict'; return true;",
+                Vec::new(),
+            )
+            .await?;
+        driver.refresh().await?;
+        wait_until(
+            &driver,
+            "const root = document.querySelector('[data-kanban-sync]'); return root?.getAttribute('data-sync-phase') === 'acknowledged' && root?.getAttribute('data-sync-pending-count') === '0'",
+        )
+        .await?;
+        let authorized = driver
+            .execute(
+                "const root = document.querySelector('[data-kanban-sync]'); return { sequence: root.getAttribute('data-sync-ack-sequence'), command: root.getAttribute('data-sync-ack-command-id'), column: root.getAttribute('data-sync-canonical-column'), pending: root.getAttribute('data-sync-pending-count') }",
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(authorized["sequence"], "1");
+        assert_eq!(authorized["command"], "auth:1");
+        assert_eq!(authorized["column"], "done");
+        assert_eq!(authorized["pending"], "0");
+        assert_eq!(command_count(&driver).await?, 0);
+        Ok(())
+    }
+    .await;
+    let quit = driver.quit().await;
+    result.and(quit)
+}
+
 #[tokio::test]
 async fn schema_upgrade_preserves_queued_order_and_local_intent() -> WebDriverResult<()> {
     // test req: sync/004 req: sync/010 req: sync/014