diff --git a/crates/libs/api/src/public_sessions.rs b/crates/libs/api/src/public_sessions.rs index acfd6994df..30bd267b2c 100644 --- a/crates/libs/api/src/public_sessions.rs +++ b/crates/libs/api/src/public_sessions.rs @@ -26,6 +26,10 @@ const SESSION_OPERATION_PROGRESS: &str = "session created"; const DEFAULT_ATTACH_TOKEN_TTL: Duration = Duration::from_secs(5 * 60); const DEFAULT_ATTACH_DEADLINE: Duration = Duration::from_secs(60); const DEFAULT_SESSION_TTL: Duration = Duration::from_secs(24 * 60 * 60); +/// Live attach tokens one session keeps. Each replay of the create mints +/// another; past this many, the replay evicts the live tokens that expire +/// first. This bounds a session's rows however often its key is replayed. +const MAX_LIVE_ATTACH_TOKENS: i64 = 16; const DATA_PLANE_TOKEN_PREFIX: &str = "dpt_"; const DEFAULT_DATA_PLANE_BASE_URL: &str = ""; const DATA_PLANE_BASE_URL_ENV: &str = "SPACE_DATA_PLANE_PUBLIC_URL"; @@ -57,6 +61,8 @@ pub enum PublicSessionStoreError { MissingIdempotencyOperation, #[error("session idempotency operation has no target session")] MissingIdempotencyTarget, + #[error("session was revoked or closed")] + SessionRevoked, #[error("database error: {0}")] Sql(#[from] sqlx::Error), } @@ -108,10 +114,29 @@ impl PublicSessionStore { let session_id = fetch_operation_target_session(&mut tx, operation_id) .await? .ok_or(PublicSessionStoreError::MissingIdempotencyTarget)?; - let token = refresh_attach_token_tx(&mut tx, session_id).await?; - let row = fetch_session_tx(&mut tx, input.org_id, input.sandbox_id, session_id) - .await? - .ok_or(PublicSessionStoreError::MissingIdempotencyTarget)?; + let token = DataPlaneToken::new(); + let Some(row) = refresh_attach_token_tx( + &mut tx, + input.org_id, + input.sandbox_id, + session_id, + &token, + ) + .await? + else { + // The refresh skips a session that has ended. One that still + // exists ended by revocation or close; a replay must not hand + // out a token for it. + return Err( + match fetch_session_tx(&mut tx, input.org_id, input.sandbox_id, session_id) + .await? + { + Some(_) => PublicSessionStoreError::SessionRevoked, + None => PublicSessionStoreError::MissingIdempotencyTarget, + }, + ); + }; + evict_attach_tokens_tx(&mut tx, session_id, &token).await?; tx.commit().await?; return PublicSessionRow::try_into_public( row, @@ -129,8 +154,11 @@ impl PublicSessionStore { let attach_deadline_at = now + DEFAULT_ATTACH_DEADLINE; let expires_at = now + DEFAULT_SESSION_TTL; + // The token row's foreign key is checked at the end of the statement, + // after the session row. let row = sqlx::query_as::<_, PublicSessionRow>(sql!( - "INSERT INTO data_plane_sessions ( + "WITH session AS ( + INSERT INTO data_plane_sessions ( session_id, org_id, actor_user_id, @@ -163,7 +191,27 @@ impl PublicSessionStore { target_path, created_at, updated_at, - expires_at" + expires_at + ), + token AS ( + INSERT INTO data_plane_session_tokens (token_hash, session_id, expires_at) + SELECT $8, session_id, $9 + FROM session + ) + SELECT session_id, + sandbox_id, + kind, + state, + reconnect_count, + last_reconnect_reason, + scopes, + target_port, + target_protocol, + target_path, + created_at, + updated_at, + expires_at + FROM session" )) .bind(&session_id_string) .bind(input.org_id) @@ -350,6 +398,14 @@ impl PublicSessionStore { .fetch_one(&mut *tx) .await?; + sqlx::query(sql!( + "DELETE FROM data_plane_session_tokens + WHERE session_id = $1" + )) + .bind(input.session_id.to_string()) + .execute(&mut *tx) + .await?; + if transitioned { sqlx::query(sql!( "INSERT INTO data_plane_org_revocation_epochs (org_id, epoch, updated_at) @@ -580,28 +636,108 @@ async fn fetch_operation_target_session( .transpose() } +/// Mint `token` for a replayed session, or `None` when the session has expired +/// or ended. The session row keeps naming the latest token for validators that +/// predate `data_plane_session_tokens`; the token row is what keeps earlier +/// tokens valid. The row lock this takes orders the replay against a +/// concurrent revoke, whose deletion of the session's tokens then includes +/// this one. async fn refresh_attach_token_tx( tx: &mut Transaction<'_, Postgres>, + org_id: OrgId, + sandbox_id: PublicSandboxId, session_id: DataPlaneSessionId, -) -> Result { - let token = DataPlaneToken::new(); + token: &DataPlaneToken, +) -> Result, sqlx::Error> { let now = OffsetDateTime::now_utc(); - let session_id_string = session_id.to_string(); - sqlx::query(sql!( - "UPDATE data_plane_sessions - SET token_hash = $2, - token_expires_at = $3, - attach_deadline_at = $4, + sqlx::query_as::<_, PublicSessionRow>(sql!( + "WITH session AS ( + UPDATE data_plane_sessions + SET token_hash = $4, + token_expires_at = $5, + attach_deadline_at = $6, updated_at = NOW() - WHERE session_id = $1" + WHERE org_id = $1 + AND sandbox_id = $2 + AND session_id = $3 + AND (expires_at IS NULL OR expires_at > NOW()) + AND revocation_status = 'active' + AND state NOT IN ('closed', 'revoked', 'failed') + RETURNING session_id, + sandbox_id, + kind, + state, + reconnect_count, + last_reconnect_reason, + scopes, + target_port, + target_protocol, + target_path, + created_at, + updated_at, + expires_at + ), + token AS ( + INSERT INTO data_plane_session_tokens (token_hash, session_id, expires_at) + SELECT $4, session_id, $5 + FROM session + ) + SELECT session_id, + sandbox_id, + kind, + state, + reconnect_count, + last_reconnect_reason, + scopes, + target_port, + target_protocol, + target_path, + created_at, + updated_at, + expires_at + FROM session" )) - .bind(&session_id_string) + .bind(org_id) + .bind(sandbox_id.as_uuid()) + .bind(session_id.to_string()) .bind(token.hash.as_str()) .bind(now + DEFAULT_ATTACH_TOKEN_TTL) .bind(now + DEFAULT_ATTACH_DEADLINE) + .fetch_optional(&mut **tx) + .await +} + +/// Delete the session's expired tokens, and its live ones beyond +/// [`MAX_LIVE_ATTACH_TOKENS`] counting `keep`, earliest expiry first. `keep` is +/// the token this replay returns and is never evicted. +async fn evict_attach_tokens_tx( + tx: &mut Transaction<'_, Postgres>, + session_id: DataPlaneSessionId, + keep: &DataPlaneToken, +) -> Result<(), sqlx::Error> { + sqlx::query(sql!( + "DELETE FROM data_plane_session_tokens + WHERE session_id = $1 + AND token_hash <> $2 + AND ( + expires_at <= NOW() + OR token_hash IN ( + SELECT token_hash + FROM data_plane_session_tokens + WHERE session_id = $1 + AND token_hash <> $2 + AND expires_at > NOW() + ORDER BY expires_at DESC, token_hash DESC + OFFSET $3 + ) + )" + )) + .bind(session_id.to_string()) + .bind(keep.hash.as_str()) + .bind(MAX_LIVE_ATTACH_TOKENS - 1) .execute(&mut **tx) .await?; - Ok(token) + Ok(()) } async fn fetch_session_tx( diff --git a/crates/libs/api/src/public_sessions/tests.rs b/crates/libs/api/src/public_sessions/tests.rs index fa57dbfa29..fb5bb875b4 100644 --- a/crates/libs/api/src/public_sessions/tests.rs +++ b/crates/libs/api/src/public_sessions/tests.rs @@ -1105,3 +1105,232 @@ async fn seed_sandbox(pool: &PgPool, org_id: OrgId, user_id: UserId) -> Result TestResult { + let org_id = OrgId::from_uuid(seed_org(pool).await?); + let actor_id = UserId::from_uuid(seed_user(pool, org_id).await?); + let sandbox_id = PublicSandboxId::from_uuid(seed_sandbox(pool, org_id, actor_id).await?); + let request = serde_json::json!({"command": [], "env": {}}); + let fingerprint = session_fingerprint_payload( + sandbox_id, + SessionKind::Terminal, + None, + None, + None, + &request, + ); + Ok(Self { + store: PublicSessionStore::with_data_plane_base_url(pool.clone(), ""), + org_id, + actor_id, + sandbox_id, + request, + fingerprint, + key: IdempotencyKey::parse(key)?, + }) + } + + async fn create(&self) -> Result { + self.store + .create_session(CreatePublicSession { + org_id: self.org_id, + actor_user_id: Some(self.actor_id), + sandbox_id: self.sandbox_id, + kind: SessionKind::Terminal, + target_port: None, + target_protocol: None, + target_path: None, + adapter_state: self.request.clone(), + fingerprint_payload: &self.fingerprint, + idempotency: CreatePublicSessionIdempotency { + method: HttpOperationMethod::Post, + path_template: OperationPathTemplate::new( + "/v1/sandboxes/{sandbox_id}/terminal-sessions", + ), + key: &self.key, + }, + }) + .await + } + + async fn revoke(&self, session_id: DataPlaneSessionId) -> TestResult { + self.store + .revoke_session(RevokePublicSession { + org_id: self.org_id, + sandbox_id: self.sandbox_id, + session_id, + reason: "user_requested", + }) + .await? + .expect("session should be revoked"); + Ok(()) + } +} + +fn attach_token_hash(session: &PublicSession) -> String { + hash_token(&session.attach.as_ref().expect("attach token").token) +} + +/// The session's token rows that have not expired, by hash. +async fn live_token_hashes( + pool: &PgPool, + session_id: DataPlaneSessionId, +) -> Result, sqlx::Error> { + let hashes = sqlx::query_scalar::<_, String>(sql!( + "SELECT token_hash + FROM data_plane_session_tokens + WHERE session_id = $1 AND expires_at > NOW()" + )) + .bind(session_id.to_string()) + .fetch_all(pool) + .await?; + Ok(hashes.into_iter().collect()) +} + +async fn token_row_count( + pool: &PgPool, + session_id: DataPlaneSessionId, +) -> Result { + sqlx::query_scalar::<_, i64>(sql!( + "SELECT COUNT(*) FROM data_plane_session_tokens WHERE session_id = $1" + )) + .bind(session_id.to_string()) + .fetch_one(pool) + .await +} + +#[tokio::test] +async fn concurrent_replays_each_add_a_token_and_keep_earlier_ones() -> TestResult { + let Some(db) = try_pg_test_db().await? else { + return Ok(()); + }; + let create = ReplayableCreate::seed(&db.pool, "concurrent-token-replays").await?; + let first = create.create().await?; + let first_hash = attach_token_hash(&first); + assert_eq!( + live_token_hashes(&db.pool, first.id).await?, + [first_hash.clone()].into() + ); + + let replays = futures_util::future::join_all((0..10).map(|_| create.create())).await; + let mut hashes = std::collections::BTreeSet::from([first_hash.clone()]); + for replay in replays { + let replay = replay?; + assert_eq!(replay.id, first.id); + hashes.insert(attach_token_hash(&replay)); + } + assert_eq!(hashes.len(), 11, "every replay mints a distinct token"); + assert_eq!(live_token_hashes(&db.pool, first.id).await?, hashes); + + // The session row still names one of them for validators that predate + // the token table. + let latest = sqlx::query_scalar::<_, Option>(sql!( + "SELECT token_hash FROM data_plane_sessions WHERE session_id = $1" + )) + .bind(first.id.to_string()) + .fetch_one(&db.pool) + .await? + .expect("session row names the latest token"); + assert!(hashes.contains(&latest)); + Ok(()) +} + +#[tokio::test] +async fn replay_prunes_expired_tokens_and_caps_live_ones() -> TestResult { + let Some(db) = try_pg_test_db().await? else { + return Ok(()); + }; + let create = ReplayableCreate::seed(&db.pool, "capped-token-replays").await?; + let first = create.create().await?; + sqlx::query(sql!( + "INSERT INTO data_plane_session_tokens (token_hash, session_id, expires_at) + VALUES ('expired-token-hash', $1, NOW() - INTERVAL '1 second')" + )) + .bind(first.id.to_string()) + .execute(&db.pool) + .await?; + + let mut minted = vec![attach_token_hash(&first)]; + for _ in 0..20 { + minted.push(attach_token_hash(&create.create().await?)); + } + + // 21 minted: the five that expire first are evicted, and the expired row + // is gone. + let live = live_token_hashes(&db.pool, first.id).await?; + let newest: std::collections::BTreeSet = minted[5..].iter().cloned().collect(); + assert_eq!(live.len(), 16); + assert_eq!(live, newest); + assert_eq!(token_row_count(&db.pool, first.id).await?, 16); + Ok(()) +} + +#[tokio::test] +async fn revoke_deletes_every_token_and_replay_is_refused() -> TestResult { + let Some(db) = try_pg_test_db().await? else { + return Ok(()); + }; + let create = ReplayableCreate::seed(&db.pool, "revoked-token-replay").await?; + let first = create.create().await?; + create.create().await?; + assert_eq!(token_row_count(&db.pool, first.id).await?, 2); + + create.revoke(first.id).await?; + assert_eq!(token_row_count(&db.pool, first.id).await?, 0); + + let replay = create + .create() + .await + .expect_err("a revoked session's replay must not mint a token"); + assert!( + matches!(replay, PublicSessionStoreError::SessionRevoked), + "{replay:?}" + ); + assert_eq!(token_row_count(&db.pool, first.id).await?, 0); + let column = sqlx::query_scalar::<_, Option>(sql!( + "SELECT token_hash FROM data_plane_sessions WHERE session_id = $1" + )) + .bind(first.id.to_string()) + .fetch_one(&db.pool) + .await?; + assert_eq!(column, None); + Ok(()) +} + +#[tokio::test] +async fn replay_of_an_ended_session_is_refused() -> TestResult { + let Some(db) = try_pg_test_db().await? else { + return Ok(()); + }; + for state in ["closed", "revoked", "failed"] { + let create = ReplayableCreate::seed(&db.pool, &format!("{state}-token-replay")).await?; + let first = create.create().await?; + sqlx::query(sql!( + "UPDATE data_plane_sessions SET state = $2 WHERE session_id = $1" + )) + .bind(first.id.to_string()) + .bind(state) + .execute(&db.pool) + .await?; + + let replay = create.create().await.expect_err(state); + assert!( + matches!(replay, PublicSessionStoreError::SessionRevoked), + "{state}: {replay:?}" + ); + assert_eq!(token_row_count(&db.pool, first.id).await?, 1, "{state}"); + } + Ok(()) +} diff --git a/crates/libs/api/src/routes/public/error.rs b/crates/libs/api/src/routes/public/error.rs index 6d93e8453d..00837f7e5a 100644 --- a/crates/libs/api/src/routes/public/error.rs +++ b/crates/libs/api/src/routes/public/error.rs @@ -373,6 +373,16 @@ impl From for PublicApiError { "idempotency key was already used with a different request", ) } + crate::public_sessions::PublicSessionStoreError::SessionRevoked => { + let mut error = Self::conflict( + "session was revoked or closed; create a new session with a new idempotency key", + ); + error.details.push(PublicErrorDetail { + key: pplx_space_public_api::CONFLICT_REASON_DETAIL_KEY.to_string(), + value: pplx_space_public_api::CONFLICT_REASON_SESSION_REVOKED.to_string(), + }); + error + } crate::public_sessions::PublicSessionStoreError::Sql(error) => Self::from(error), other => { tracing::error!(error = %other, "public session store error"); diff --git a/crates/libs/api/src/routes/public/error/tests.rs b/crates/libs/api/src/routes/public/error/tests.rs index 5db273eaf1..a93690480d 100644 --- a/crates/libs/api/src/routes/public/error/tests.rs +++ b/crates/libs/api/src/routes/public/error/tests.rs @@ -350,3 +350,26 @@ async fn unhandled_429_keeps_its_retry_after_when_normalized() { assert!(envelope.retryable); assert_eq!(envelope.request_id.as_deref(), Some(REQUEST_ID)); } + +/// Replaying the create of a revoked or closed session answers the existing +/// `conflict` code, distinguished by its detail rather than a new code. +#[tokio::test] +async fn revoked_session_replay_is_a_non_retryable_conflict_with_a_reason() { + use axum::response::IntoResponse; + + let error = + PublicApiError::from(crate::public_sessions::PublicSessionStoreError::SessionRevoked); + let response = error.into_response(); + assert_eq!(response.status(), StatusCode::CONFLICT); + let envelope: PublicErrorEnvelope = + serde_json::from_slice(&response.into_body().collect().await.unwrap().to_bytes()).unwrap(); + assert_eq!(envelope.code, PublicErrorCode::Conflict); + assert!(!envelope.retryable); + assert_eq!( + envelope.details, + vec![pplx_space_public_api::PublicErrorDetail { + key: "reason".to_string(), + value: "session_revoked".to_string(), + }] + ); +} diff --git a/crates/libs/internal/api/server/src/data_plane_sessions/route_rows.rs b/crates/libs/internal/api/server/src/data_plane_sessions/route_rows.rs index b5f09a4586..5b80a0f924 100644 --- a/crates/libs/internal/api/server/src/data_plane_sessions/route_rows.rs +++ b/crates/libs/internal/api/server/src/data_plane_sessions/route_rows.rs @@ -8,7 +8,10 @@ pub(super) async fn fetch_route_row( pool: &sqlx::PgPool, session_id: &str, protocol_session_id: Option<&str>, + token_sha256: Option<&str>, ) -> Result { + // A revoke from a server that predates `data_plane_session_tokens` clears + // only `d.token_hash`, so a token row counts only while that is set. sqlx::query_as::<_, SessionRouteRow>(sql!( "SELECT d.session_id, d.protocol_session_id, @@ -31,6 +34,11 @@ pub(super) async fn fetch_route_row( d.scopes, d.token_hash, d.token_expires_at, + (SELECT t.expires_at + FROM data_plane_session_tokens t + WHERE t.token_hash = $3 + AND t.session_id = d.session_id + AND d.token_hash IS NOT NULL) AS presented_token_expires_at, s.lifecycle AS sandbox_lifecycle, COALESCE(direct_route.current_node_id, s.active_node_id) AS sandbox_active_node_id, @@ -61,6 +69,7 @@ pub(super) async fn fetch_route_row( )) .bind(session_id) .bind(protocol_session_id) + .bind(token_sha256) .fetch_optional(pool) .await .map_err(|error| Status::internal(format!("resolve data-plane session route: {error}")))? diff --git a/crates/libs/internal/api/server/src/data_plane_sessions/session_route.rs b/crates/libs/internal/api/server/src/data_plane_sessions/session_route.rs index b624359bf7..64776e339b 100644 --- a/crates/libs/internal/api/server/src/data_plane_sessions/session_route.rs +++ b/crates/libs/internal/api/server/src/data_plane_sessions/session_route.rs @@ -26,6 +26,11 @@ pub(super) struct SessionRouteRow { pub(super) scopes: Vec, pub(super) token_hash: Option, pub(super) token_expires_at: Option, + /// Expiry of the presented attach token's `data_plane_session_tokens` row, + /// `None` when it has none. Only [`fetch_route_row`] reads it; authorization + /// runs on that row, before any route update. + #[sqlx(default)] + pub(super) presented_token_expires_at: Option, pub(super) sandbox_active_node_id: Option, pub(super) sandbox_lifecycle: SandboxLifecycle, pub(super) node_address: Option, @@ -53,8 +58,14 @@ pub(crate) async fn resolve_session_route( let deadline = tokio::time::Instant::now() + budget; let mut activated = false; loop { - let mut row = fetch_route_row(pool, &session_id, protocol_session_id.as_deref()).await?; - validate_attach_authorization( + let mut row = fetch_route_row( + pool, + &session_id, + protocol_session_id.as_deref(), + token_sha256.as_deref(), + ) + .await?; + let token_expires_at = validate_attach_authorization( &row, refresh_authority, token_sha256.as_deref(), @@ -64,7 +75,7 @@ pub(crate) async fn resolve_session_route( row.kind.parse::(), Ok(SessionKind::Port | SessionKind::Protocol) ) { - validate_ingress_session(&row)?; + validate_ingress_session(&row, token_expires_at)?; Some( pplx_space_common_svc::ingress_admission::IngressPreparationPermit::acquire(Some( row.sandbox_id, @@ -308,10 +319,13 @@ pub(super) fn data_plane_session_transport( } } -pub(super) fn validate_ingress_session(row: &SessionRouteRow) -> Result<(), Status> { +pub(super) fn validate_ingress_session( + row: &SessionRouteRow, + token_expires_at: Option, +) -> Result<(), Status> { validate_ingress_session_access( &row.revocation_status, - row.token_expires_at, + token_expires_at, OffsetDateTime::now_utc(), ) } @@ -330,19 +344,21 @@ pub(super) fn validate_ingress_session_access( Ok(()) } +/// Authorize an attach, returning the expiry that bounds it: the presented +/// token's, or for a tokenless edge refresh the session's latest token's. pub(super) fn validate_attach_authorization( row: &SessionRouteRow, refresh_authority: Option, token_sha256: Option<&str>, required_scope: Option<&str>, -) -> Result<(), Status> { +) -> Result, Status> { if token_sha256.is_none() && required_scope.is_none() { // The data-plane edge refreshes the route of a session it already // attached without re-sending the attach secret. Any other caller // holding only a session ID must prove the attach, or the ID alone // would mint a route token for another tenant's session. return match refresh_authority { - Some(DataPlaneSessionRefreshAuthority::Granted) => Ok(()), + Some(DataPlaneSessionRefreshAuthority::Granted) => Ok(row.token_expires_at), _ => Err(Status::unauthenticated( "data-plane attach token is required", )), @@ -353,22 +369,31 @@ pub(super) fn validate_attach_authorization( .map(str::trim) .filter(|value| !value.is_empty()) .ok_or_else(|| Status::unauthenticated("data-plane attach token is required"))?; - let stored_hash = row - .token_hash - .as_deref() - .map(str::trim) - .filter(|value| !value.is_empty()) - .ok_or_else(|| Status::unauthenticated("data-plane session has no attach token"))?; - if stored_hash - .as_bytes() - .ct_eq(token_sha256.as_bytes()) - .unwrap_u8() - != 1 - { - return Err(Status::unauthenticated("invalid data-plane attach token")); - } + // A token with a row in `data_plane_session_tokens` carries its own expiry. + // Any other token must be the one the session row names: the latest, and + // the only one a server that predates the table minted. + let token_expires_at = match row.presented_token_expires_at { + Some(expires_at) => Some(expires_at), + None => { + let stored_hash = row + .token_hash + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + .ok_or_else(|| Status::unauthenticated("data-plane session has no attach token"))?; + if stored_hash + .as_bytes() + .ct_eq(token_sha256.as_bytes()) + .unwrap_u8() + != 1 + { + return Err(Status::unauthenticated("invalid data-plane attach token")); + } + row.token_expires_at + } + }; - match row.token_expires_at { + match token_expires_at { Some(expires_at) if expires_at > OffsetDateTime::now_utc() => {} Some(_) => return Err(Status::unauthenticated("data-plane attach token expired")), None => { @@ -386,5 +411,5 @@ pub(super) fn validate_attach_authorization( )); } - Ok(()) + Ok(token_expires_at) } diff --git a/crates/libs/internal/api/server/src/tests.rs b/crates/libs/internal/api/server/src/tests.rs index 4e05a50a41..49be0b0c7e 100644 --- a/crates/libs/internal/api/server/src/tests.rs +++ b/crates/libs/internal/api/server/src/tests.rs @@ -2591,6 +2591,98 @@ async fn data_plane_route_resolve_validates_token_and_scope() -> TestResult { Ok(()) } +/// Every live token a session's create or replays minted attaches, each until +/// its own expiry; a token with no row still attaches when the session row +/// names it, as one minted by a server that predates the token table does. +#[tokio::test(flavor = "current_thread")] +async fn data_plane_route_resolve_accepts_each_live_session_token() -> TestResult { + let Some(fixture) = cluster_fixture().await? else { + return Ok(()); + }; + let node_id = NodeId::new(); + let sandbox_id = SandboxId::new(); + let session_id = format!("term_{}", Uuid::now_v7()); + let other_session_id = format!("term_{}", Uuid::now_v7()); + + insert_data_plane_worker_node(&fixture.service.pool, node_id, "10.0.0.32").await?; + insert_data_plane_active_sandbox(&fixture.service.pool, sandbox_id, node_id).await?; + insert_active_execution(&fixture.service.pool, sandbox_id, "10.80.0.32", 32).await?; + insert_data_plane_session(&fixture.service.pool, &session_id, sandbox_id).await?; + insert_data_plane_session(&fixture.service.pool, &other_session_id, sandbox_id).await?; + sqlx::query( + "UPDATE data_plane_sessions + SET token_hash = 'latest-token', + token_expires_at = NOW() + ('5 minutes')::INTERVAL, + scopes = ARRAY['terminal:connect']", + ) + .execute(&fixture.service.pool) + .await?; + sqlx::query( + "INSERT INTO data_plane_session_tokens (token_hash, session_id, expires_at) + VALUES ('latest-token', $1, NOW() + ('5 minutes')::INTERVAL), + ('earlier-token', $1, NOW() + ('2 minutes')::INTERVAL), + ('expired-token', $1, NOW() - ('1 second')::INTERVAL), + ('other-session-token', $2, NOW() + ('5 minutes')::INTERVAL)", + ) + .bind(&session_id) + .bind(&other_session_id) + .execute(&fixture.service.pool) + .await?; + + let resolve = |session_id: &str, token: &str| { + fixture + .service + .resolve_data_plane_session_route(data_plane_request( + ResolveDataPlaneSessionRouteRequest { + session_id: session_id.to_string(), + protocol_session_id: None, + token_sha256: Some(token.to_string()), + required_scope: Some("terminal:connect".to_string()), + requested_transport: None, + }, + )) + }; + + resolve(&session_id, "latest-token").await?; + resolve(&session_id, "earlier-token").await?; + for token in ["expired-token", "other-session-token", "unknown-token"] { + let error = resolve(&session_id, token) + .await + .expect_err("token must not attach"); + assert_eq!(error.code(), Code::Unauthenticated, "{token}"); + } + + // The session row's token attaches without a row of its own. + sqlx::query( + "UPDATE data_plane_sessions + SET token_hash = 'column-only-token' + WHERE session_id = $1", + ) + .bind(&session_id) + .execute(&fixture.service.pool) + .await?; + resolve(&session_id, "column-only-token").await?; + resolve(&session_id, "earlier-token").await?; + + // A revoke by a server that predates the token table clears only the + // session row's token; that alone must end every token. + sqlx::query( + "UPDATE data_plane_sessions + SET token_hash = NULL, token_expires_at = NOW() + WHERE session_id = $1", + ) + .bind(&session_id) + .execute(&fixture.service.pool) + .await?; + for token in ["latest-token", "earlier-token", "column-only-token"] { + let error = resolve(&session_id, token) + .await + .expect_err("revoked session token must not attach"); + assert_eq!(error.code(), Code::Unauthenticated, "{token}"); + } + Ok(()) +} + #[tokio::test(flavor = "current_thread")] async fn ecw_route_resolve_allows_envd_with_api_key() -> TestResult { if !pplx_space_test_env::enter_test_process(&[ diff --git a/crates/libs/public/api/src/lib.rs b/crates/libs/public/api/src/lib.rs index a24f4575ed..3c8946a902 100644 --- a/crates/libs/public/api/src/lib.rs +++ b/crates/libs/public/api/src/lib.rs @@ -2020,6 +2020,14 @@ pub const NOT_FOUND_RESOURCE_DETAIL_KEY: &str = "resource"; /// sandbox and another org's sandbox stay indistinguishable. pub const NOT_FOUND_RESOURCE_PATH: &str = "path"; +/// [`PublicErrorDetail::key`] naming why a `conflict` was refused. +pub const CONFLICT_REASON_DETAIL_KEY: &str = "reason"; + +/// [`CONFLICT_REASON_DETAIL_KEY`] value of a `conflict` for replaying a +/// session create whose session was revoked or closed. Replaying cannot renew +/// its token; create a new session with a new idempotency key. +pub const CONFLICT_REASON_SESSION_REVOKED: &str = "session_revoked"; + open_wire_enum!(DataPlaneHttpErrorCode, "data-plane HTTP error code", { InvalidRequest => "invalid_request", Unauthorized => "unauthorized", diff --git a/docs/api/index.md b/docs/api/index.md index 1e28be0802..1707b7cf25 100644 --- a/docs/api/index.md +++ b/docs/api/index.md @@ -30,6 +30,14 @@ Ready to run code? Follow [Your first sandbox](../first-sandbox.md) for a comple - **Create and attach in one request** by sending `Accept: text/event-stream` with the create. The response is the event stream, preceded by a `command` frame whose `command` field is the handle. If the server answers with the JSON handle instead, attach as usual. - **Save output as it arrives.** Streams are live, with no replay. Check `process_exited`; a closed connection alone is not success. See [command output](../first-sandbox.md#read-command-output). +## Renew session tokens + +A command's session token, and the token of any other data-plane session, is valid for 5 minutes, until its `attach.expires_at`. + +- **Replay the create to get another token.** Sending the same method, path, body, and `Idempotency-Key` again returns the same session with an additional 5-minute token. Earlier tokens stay valid until their own `expires_at`, so a renewal or a concurrent retry does not invalidate a token still in use. A session keeps at most 16 unexpired tokens; past that, a replay revokes the ones that expire first. +- **An attached stream outlives its token.** It is not closed when the token expires. Reconnecting needs an unexpired token, and so does every request through a port or protocol session. +- **Renewal ends with the session**, 24 hours after it was created. Revoking a session ends all of its tokens. Replaying the create of a revoked or closed session returns `409` `conflict` with the detail `{"key": "reason", "value": "session_revoked"}` and no token; create a new session with a new key instead. + ## Retry safely Error samples are illustrative, not exhaustive. Messages can change, and the same HTTP status can have different codes or retryability. Follow the returned `retryable` flag and `Retry-After`, not the sample. diff --git a/docs/native-api-hardening.md b/docs/native-api-hardening.md index 384e6a471f..bf9b19fb3c 100644 --- a/docs/native-api-hardening.md +++ b/docs/native-api-hardening.md @@ -57,7 +57,7 @@ the ordinary smoke job logs. |---|---|---| | Contract | Command validation, changed-request idempotency conflicts, token scope, tenant isolation | Included | | Contract | Completed DELETE replay, permanent startup errors | Included | -| Proposed contract | Overlapping replay credential validity | Ignored; explicit invocation only | +| Proposed contract | Earlier credentials stay valid after an idempotent replay | Included | | Admission | Zero-vCPU rejection before accepting an operation | Included on Kind with bounded cleanup | | Concurrency | Same-key create and command fanout, independent commands, pause/wake contention, slow-reader output integrity | Included | | Recovery | Queued create across controller outage/API restart; original command session across data-plane restart | Included on Kind only | @@ -99,29 +99,29 @@ Named development benches require `SPACE_SMOKE_KIND_CLUSTER_UID` to match the live `kube-system` namespace UID before privileged fixture operations. The standard `kind-space-local` setup retains its existing isolation checks. -## Credential replay contract (deferred) - -The current API rotates attachment credentials on an idempotent replay. -`native_concurrency.command_idempotency` remains required: same-key requests -share session and process identity, execute their side effect once, and return -an attachable fresh credential after concurrent requests finish. - -`native_contract.command_replay_credentials` additionally expects the previous -credential to remain valid. That stronger guarantee is not agreed. Its reproducer -is explicitly `ignored`, appears as `IGNORED` in suite logs, and is excluded from -normal selection and the quarantine defect backlog. It is not fixed or counted -as a passing scenario. Run its unchanged assertions explicitly with: - -```bash -SPACE_SMOKE_BACKEND=kind SPACE_SMOKE_API_SURFACE=native \ - uv run --frozen --python 3.11 -m space_smoke run native_contract.command_replay_credentials -``` - -The [review decision](https://github.com/perplexityai/SPACE/pull/79#pullrequestreview-5423918313) -defers overlap. The [separate storage proposal](https://github.com/perplexityai/SPACE/pull/91) -is not a prerequisite for the smoke repairs. Before adopting an implementation, -resolve replay versus explicit credential renewal, validity of earlier responses, -expiry, revocation, issuance bounds, and deployment compatibility together. +## Credential replay contract (proposed) + +Whether an idempotent replay of a command create should also renew its +attachment credential is not yet agreed. The +[review of #79](https://github.com/perplexityai/SPACE/pull/79#pullrequestreview-5423918313) +and the [review of #91](https://github.com/perplexityai/SPACE/pull/91#pullrequestreview-5423917990) +deferred it: command-create replay is to be defined separately from credential +renewal, including expiry, revocation, and whether overlapping validity is +promised, before choosing the smallest implementation. + +The behavior described here and in +[session token renewal](api/index.md#renew-session-tokens) is one proposal +pending that agreement. A replay returns the same session with an additional +attachment credential, credentials returned earlier stay valid until their own +expiry, at most 16 stay valid per session, and revocation ends all of them. +Alternatives are an explicit credential-renew endpoint that leaves replay +unchanged, and a single `previous_token_hash` grace column that keeps only the +credential before the latest one valid. + +`native_contract.command_replay_credentials` checks that the original +credential still attaches after a replay. `native_concurrency.command_idempotency` +checks that same-key requests share session and process identity, execute their +side effect once, and return an attachable credential. ## Concurrency and recovery controls diff --git a/space_smoke/scenarios/native_concurrency_command_idempotency.py b/space_smoke/scenarios/native_concurrency_command_idempotency.py index c55e66f771..faa0ed521d 100644 --- a/space_smoke/scenarios/native_concurrency_command_idempotency.py +++ b/space_smoke/scenarios/native_concurrency_command_idempotency.py @@ -44,9 +44,9 @@ def submit(index: int, trace: CallerTrace, deadline: float) -> Mapping[str, obje outcomes = run_fanout(test, settings, "command_replay_fanout", submit) identities = {(item.value["session_id"], item.value["process_id"]) for item in outcomes if item.value} test.context.verify(len(identities) == 1, "same-key command retries produced multiple sessions or processes") - # The current contract rotates the attachment token on replay. Fetch a fresh - # credential after concurrent POSTs drain; this case proves execution count, - # not validity of credentials returned by overlapping retries. + # Fetch a fresh credential after concurrent POSTs drain; this case proves + # execution count. native_contract.command_replay_credentials covers the + # validity of credentials returned by earlier replays. response = test.request("POST", path, body=body, idempotency_key=key) test.context.verify(response.status == 201, "final command replay failed") test.context.verify((response.body.get("session_id"), response.body.get("process_id")) in identities, diff --git a/space_smoke/scenarios/native_contract_command_replay_credentials.py b/space_smoke/scenarios/native_contract_command_replay_credentials.py index da5e98d186..7f4195dfe1 100644 --- a/space_smoke/scenarios/native_contract_command_replay_credentials.py +++ b/space_smoke/scenarios/native_contract_command_replay_credentials.py @@ -2,14 +2,7 @@ from space_smoke.registry import scenario from space_smoke.native_hardening import NativeHardeningFixture -@scenario( - "native_contract.command_replay_credentials", - api_surfaces=("native",), - ignored=( - "Overlapping credential validity is not an agreed API guarantee; " - "decision: docs/native-api-hardening.md#credential-replay-contract-deferred" - ), -) +@scenario("native_contract.command_replay_credentials", api_surfaces=("native",)) def command_replay_credentials(test: NativeHardeningFixture) -> None: """A command retry preserves its session and reports any credential invalidation.""" sid = test.create_ready()