Implement replay-safe browser synchronization

crates/remora-api/src/public.rs

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use std::{collections::HashSet, time::Duration};
 
 use actix_web::{
     HttpRequest, HttpResponse, ResponseError,
@@ -51,6 +51,8 @@ pub struct PublicApiConfig {
     pub event_retention: Duration,
     pub event_poll_interval: Duration,
     pub event_heartbeat_interval: Duration,
+    pub event_ticket_ttl: Duration,
+    pub event_allowed_origins: Vec<String>,
 }
 
 #[derive(Clone)]
@@ -68,6 +70,8 @@ pub struct PublicApiState {
     event_retention: Duration,
     event_poll_interval: Duration,
     event_heartbeat_interval: Duration,
+    event_ticket_ttl: Duration,
+    event_allowed_origins: HashSet<String>,
 }
 
 pub struct PublicApiServices {
@@ -93,9 +97,28 @@ impl PublicApiState {
         if config.event_retention.is_zero()
             || config.event_poll_interval.is_zero()
             || config.event_heartbeat_interval.is_zero()
+            || config.event_ticket_ttl.is_zero()
         {
             return Err(DurableError::Config(
-                "event retention and intervals must be positive".into(),
+                "event retention, intervals, and ticket expiry must be positive".into(),
+            ));
+        }
+        if config.event_ticket_ttl > Duration::from_secs(60) {
+            return Err(DurableError::Config(
+                "event ticket expiry must not exceed 60 seconds".into(),
+            ));
+        }
+        let event_allowed_origins = config
+            .event_allowed_origins
+            .into_iter()
+            .collect::<HashSet<_>>();
+        if event_allowed_origins.is_empty()
+            || event_allowed_origins
+                .iter()
+                .any(|origin| origin.is_empty() || origin.trim() != origin)
+        {
+            return Err(DurableError::Config(
+                "at least one exact event origin is required".into(),
             ));
         }
         Ok(Self {
@@ -112,6 +135,8 @@ impl PublicApiState {
             event_retention: config.event_retention,
             event_poll_interval: config.event_poll_interval,
             event_heartbeat_interval: config.event_heartbeat_interval,
+            event_ticket_ttl: config.event_ticket_ttl,
+            event_allowed_origins,
         })
     }
 
@@ -193,6 +218,7 @@ pub fn configure_public_api(config: &mut web::ServiceConfig) {
                 "/deletions/{operation_id}",
                 web::get().to(get_deletion_operation),
             )
+            .route("/event-tickets", web::post().to(events::issue_ticket))
             .route("/events", web::get().to(events::connect)),
     );
 }

crates/remora-api/src/public/events.rs

@@ -1,16 +1,19 @@
-use actix_web::{HttpRequest, HttpResponse, web};
+use actix_web::{HttpRequest, HttpResponse, http::header, web};
 use futures_util::StreamExt as _;
 use remora_contracts::{
     CursorDto, DateTimeDto, DeletionChangedDto, DeletionStateDto, EventAttemptDto, EventFrameDto,
-    EventProgressDto, HeartbeatDto, JobChangedDto, NullableDto, PositiveIntegerDto,
+    EventProgressDto, EventTicketDto, HeartbeatDto, JobChangedDto, NullableDto, PositiveIntegerDto,
     ResyncReasonDto, ResyncRequiredDto, SafeIntegerDto, SchemaVersionDto,
 };
-use remora_domain::{JobLifecycle, JobPhase, OpaqueId};
+use remora_domain::{
+    BucketName, JobLifecycle, JobPhase, OpaqueId, StorageScope, TargetPrefix, TargetStoreId,
+};
 use remora_durable::{DurableError, KeysetCursor};
 use remora_identity::{ManagementOperation, Principal};
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use sqlx::{FromRow, PgPool};
+use sha2::{Digest as _, Sha256};
+use sqlx::{FromRow, PgPool, types::Json};
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 
 use super::{
@@ -24,6 +27,60 @@ const EVENT_BATCH: i64 = 100;
 #[derive(Deserialize)]
 pub(super) struct EventQuery {
     cursor: Option<String>,
+    ticket: Option<String>,
+}
+
+#[derive(Deserialize, Serialize)]
+struct TicketPrincipal {
+    id: String,
+    issuer: String,
+    subject: String,
+    administrator: bool,
+    scopes: Vec<TicketScope>,
+}
+
+#[derive(Deserialize, Serialize)]
+struct TicketScope {
+    store: String,
+    bucket: String,
+    prefix: String,
+}
+
+pub(super) async fn issue_ticket(
+    state: web::Data<PublicApiState>,
+    request: HttpRequest,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&request).await?;
+    let origin = allowed_origin(&state, &request)?;
+    let ticket = random_ticket();
+    let digest = ticket_digest(ticket.as_str());
+    let mut tx = state.pool.begin().await?;
+    sqlx::query("DELETE FROM live_event_tickets WHERE expires_at <= now()")
+        .execute(tx.as_mut())
+        .await?;
+    let expires_at: OffsetDateTime = sqlx::query_scalar(
+        "INSERT INTO live_event_tickets (
+            ticket_digest, principal, origin, created_at, expires_at
+         ) VALUES (
+            $1, $2, $3, now(),
+            now() + $4 * interval '1 second'
+         )
+         RETURNING expires_at",
+    )
+    .bind(digest.as_slice())
+    .bind(Json(ticket_principal_projection(&principal)))
+    .bind(&origin)
+    .bind(state.event_ticket_ttl.as_secs_f64())
+    .fetch_one(tx.as_mut())
+    .await?;
+    tx.commit().await?;
+    Ok(HttpResponse::Created()
+        .insert_header((header::CACHE_CONTROL, "no-store"))
+        .insert_header((header::PRAGMA, "no-cache"))
+        .json(EventTicketDto {
+            ticket,
+            expires_at: event_time(expires_at)?,
+        }))
 }
 
 pub(super) async fn connect(
@@ -32,7 +89,13 @@ pub(super) async fn connect(
     body: web::Payload,
     query: web::Query<EventQuery>,
 ) -> Result<HttpResponse, PublicApiError> {
-    let principal = state.principal(&request).await?;
+    let bearer = request.headers().contains_key(header::AUTHORIZATION);
+    let principal = match (bearer, query.ticket.as_deref()) {
+        (true, None) => state.principal(&request).await?,
+        (false, Some(ticket)) => redeem_ticket(&state, &request, ticket).await?,
+        (false, None) => return Err(PublicApiError::unauthenticated()),
+        (true, Some(_)) => return Err(PublicApiError::invalid_request()),
+    };
     let now = OffsetDateTime::now_utc();
     let retention =
         time::Duration::try_from(state.event_retention).map_err(|_| PublicApiError::internal())?;
@@ -54,6 +117,94 @@ pub(super) async fn connect(
     Ok(response)
 }
 
+fn allowed_origin(state: &PublicApiState, request: &HttpRequest) -> Result<String, PublicApiError> {
+    let mut origins = request.headers().get_all(header::ORIGIN);
+    let origin = origins
+        .next()
+        .and_then(|value| value.to_str().ok())
+        .ok_or_else(PublicApiError::forbidden)?;
+    if origins.next().is_some() || !state.event_allowed_origins.contains(origin) {
+        return Err(PublicApiError::forbidden());
+    }
+    Ok(origin.to_owned())
+}
+
+async fn redeem_ticket(
+    state: &PublicApiState,
+    request: &HttpRequest,
+    ticket: &str,
+) -> Result<Principal, PublicApiError> {
+    let origin = allowed_origin(state, request)?;
+    let digest = ticket_digest(ticket);
+    let projection: Option<Json<TicketPrincipal>> = sqlx::query_scalar(
+        "DELETE FROM live_event_tickets
+         WHERE ticket_digest = $1 AND origin = $2 AND expires_at > now()
+         RETURNING principal",
+    )
+    .bind(digest.as_slice())
+    .bind(origin)
+    .fetch_optional(&state.pool)
+    .await?;
+    projection
+        .ok_or_else(PublicApiError::unauthenticated)
+        .and_then(|projection| ticket_principal(projection.0))
+}
+
+fn ticket_principal_projection(principal: &Principal) -> TicketPrincipal {
+    TicketPrincipal {
+        id: principal.id.as_str().to_owned(),
+        issuer: principal.issuer.clone(),
+        subject: principal.subject.clone(),
+        administrator: principal.administrator,
+        scopes: principal
+            .scopes
+            .iter()
+            .map(|scope| TicketScope {
+                store: scope.store.as_str().to_owned(),
+                bucket: scope.bucket.as_str().to_owned(),
+                prefix: scope.prefix.canonical(),
+            })
+            .collect(),
+    }
+}
+
+fn ticket_principal(projection: TicketPrincipal) -> Result<Principal, PublicApiError> {
+    let scopes = projection
+        .scopes
+        .into_iter()
+        .map(|scope| {
+            Ok(StorageScope {
+                store: TargetStoreId::parse(scope.store).map_err(|_| PublicApiError::internal())?,
+                bucket: BucketName::parse(scope.bucket).map_err(|_| PublicApiError::internal())?,
+                prefix: TargetPrefix::parse(&scope.prefix, std::iter::empty())
+                    .map_err(|_| PublicApiError::internal())?,
+            })
+        })
+        .collect::<Result<Vec<_>, PublicApiError>>()?;
+    Ok(Principal {
+        id: OpaqueId::parse(projection.id).map_err(|_| PublicApiError::internal())?,
+        issuer: projection.issuer,
+        subject: projection.subject,
+        administrator: projection.administrator,
+        scopes,
+    })
+}
+
+fn random_ticket() -> OpaqueId {
+    let mut bytes = [0_u8; 32];
+    getrandom::fill(&mut bytes).expect("operating system randomness is required");
+    let mut value = String::from("t");
+    for byte in bytes {
+        use std::fmt::Write as _;
+        write!(value, "{byte:02x}").expect("writing to String cannot fail");
+    }
+    OpaqueId::parse(value).expect("random event ticket is valid")
+}
+
+fn ticket_digest(ticket: &str) -> [u8; 32] {
+    Sha256::digest(ticket.as_bytes()).into()
+}
+
 pub(super) async fn prune_history(
     pool: &PgPool,
     before: OffsetDateTime,