From 652743e8395acd6db213e40f0f72c49fb96e7866 Mon Sep 17 00:00:00 2001 From: NancyKwongPerplexity Date: Wed, 21 Oct 2026 18:34:38 +0000 Subject: [PATCH 1/2] api: keep earlier attach tokens valid when a session create is replayed A data-plane session stored one attach token, so replaying its create with the same idempotency key overwrote the token every earlier response returned. Concurrent same-key replays left all but one caller with a token that failed with a non-retryable 401, and renewing a port session's token by replay could 401 requests still in flight on the previous one. Replay also ignored the session's status and returned a token for a revoked session. Create and replay now each add a row to data_plane_session_tokens with its own five-minute expiry, in the statement that writes the session and under the existing idempotency lock. Replay deletes expired rows and keeps at most 16 unexpired tokens per session, evicting those that expire first but never the one it returns. The session row still names the latest token so validators that predate the table keep working. Route resolution uses the presented token's row and its own expiry for attach and for each port or protocol request, and falls back to the session row's token for tokens minted before the table. A row counts only while the session row still names a token, so a revoke by an older server still ends every token. Revoking deletes the session's rows. Replaying the create of a revoked or closed session returns the existing conflict code with the detail reason=session_revoked instead of a token. The replay-credential smoke scenario, parked until this contract was decided, runs in normal selection again. --- crates/libs/api/src/public_sessions.rs | 170 +++++++++++-- crates/libs/api/src/public_sessions/tests.rs | 229 ++++++++++++++++++ crates/libs/api/src/routes/public/error.rs | 10 + .../libs/api/src/routes/public/error/tests.rs | 23 ++ .../src/data_plane_sessions/route_rows.rs | 9 + .../src/data_plane_sessions/session_route.rs | 71 ++++-- crates/libs/internal/api/server/src/tests.rs | 92 +++++++ crates/libs/public/api/src/lib.rs | 8 + docs/api/index.md | 8 + docs/native-api-hardening.md | 33 +-- .../native_concurrency_command_idempotency.py | 6 +- ...ive_contract_command_replay_credentials.py | 9 +- 12 files changed, 595 insertions(+), 73 deletions(-) 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..f42a1d697d 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 | +| 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,18 @@ 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) +## Credential replay contract -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. +An idempotent replay of a session create returns the same session with an +additional attachment credential. Credentials returned earlier stay valid until +their own expiry; revocation ends all of them. The API reference's +[session token renewal](api/index.md#renew-session-tokens) section states the +expiry, issuance bound, and revoked-session behavior. -`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. +`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() From 623c5ae797c598ec8564375704da3cf54cff6de9 Mon Sep 17 00:00:00 2001 From: NancyKwongPerplexity Date: Wed, 21 Oct 2026 19:34:31 +0000 Subject: [PATCH 2/2] docs: describe the replay credential contract as a proposal Overlapping credential validity after a replay is not yet agreed. Mark the section and its coverage row as a proposed contract, link the reviews that deferred the decision, and list the alternatives: an explicit renew endpoint, or a single previous-token grace column. --- docs/native-api-hardening.md | 27 +++++++++++++++++++-------- 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/docs/native-api-hardening.md b/docs/native-api-hardening.md index f42a1d697d..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 | -| Contract | Earlier credentials stay valid after an idempotent replay | Included | +| 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,13 +99,24 @@ 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 - -An idempotent replay of a session create returns the same session with an -additional attachment credential. Credentials returned earlier stay valid until -their own expiry; revocation ends all of them. The API reference's -[session token renewal](api/index.md#renew-session-tokens) section states the -expiry, issuance bound, and revoked-session behavior. +## 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`