diff --git a/Cargo.lock b/Cargo.lock index 78c6f5d255..4caf98a6e5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4997,6 +4997,7 @@ dependencies = [ "pplx-space-sandbox-store", "pplx-space-sandbox-teardown", "pplx-space-sbx-mgr-client", + "pplx-space-snapshot-store", "pplx-space-template-mgr-client", "pplx-space-test-env", "prost", diff --git a/crates/libs/api/Cargo.toml b/crates/libs/api/Cargo.toml index 2074013e42..032717ca21 100644 --- a/crates/libs/api/Cargo.toml +++ b/crates/libs/api/Cargo.toml @@ -95,6 +95,7 @@ pplx-space-common-work-coalescer = { workspace = true, optional = true } pplx-space-cluster-pool = { workspace = true } pplx-space-net-common = { workspace = true, features = ["serde"] } pplx-space-sandbox-portability = { workspace = true } +pplx-space-snapshot-store = { workspace = true } pplx-space-sandbox-store = { workspace = true } pplx-space-sandbox-teardown = { workspace = true } pplx-space-sbx-mgr-client = { workspace = true } diff --git a/crates/libs/api/src/routes/admin/backup_recovery.rs b/crates/libs/api/src/routes/admin/backup_recovery.rs new file mode 100644 index 0000000000..8beb750542 --- /dev/null +++ b/crates/libs/api/src/routes/admin/backup_recovery.rs @@ -0,0 +1,208 @@ +//! Operator authority for one verified retained snapshot. Creating a grant +//! preserves bytes; it does not enqueue work, wake a sandbox, or select a restore. +use super::audit::{self, AdminSandboxAuditEntry}; +use super::support::{api_error, require_managing_org_admin}; +use crate::{middleware::auth::AuthUser, routes::AppState}; +use axum::{ + Extension, Json, + extract::{Path, State}, + http::StatusCode, + response::{IntoResponse, Response}, +}; +use pplx_space_common_core::types::{ExecutionGeneration, K8sNodeUid, LocalArtifactId, NodeId}; +use pplx_space_public_api::PublicSandboxId; +use pplx_space_snapshot_store::recovery::{self, RecoveryGrantId, RecoveryManifest}; +use serde::Deserialize; + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(super) struct Approval { + local_artifact_id: LocalArtifactId, + execution_generation: ExecutionGeneration, + uploader_node_id: NodeId, + uploader_node_uid: K8sNodeUid, + captured_at: String, + capture_evidence: recovery::CaptureTimeEvidence, + expires_at: String, + reason: String, + manifest: RecoveryManifest, +} + +fn timestamp(value: &str) -> Result { + let value: jiff::Timestamp = value + .parse() + .map_err(|_| "expected an absolute capture/expiry timestamp")?; + time::OffsetDateTime::from_unix_timestamp_nanos(value.as_nanosecond()) + .map_err(|_| "timestamp out of range") +} +fn store_error(error: recovery::RecoveryError) -> Response { + match error { + recovery::RecoveryError::Refused(message) => { + api_error(StatusCode::CONFLICT, message).into_response() + } + recovery::RecoveryError::Database(error) => super::support::db_error(error).into_response(), + other => { + tracing::error!(error=%other,"backup recovery grant transaction failed"); + api_error( + StatusCode::INTERNAL_SERVER_ERROR, + "backup recovery transaction failed", + ) + .into_response() + } + } +} + +async fn audit_org( + conn: &mut sqlx::PgConnection, + sandbox: PublicSandboxId, +) -> Result> { + match pplx_space_sandbox_store::get(conn, sandbox.into()).await { + Ok(Some(row)) => Ok(row.org_id), + Ok(None) => Err(Box::new( + api_error(StatusCode::NOT_FOUND, "sandbox not found").into_response(), + )), + Err(error) => { + tracing::error!(%error, "failed to resolve recovery audit scope"); + Err(Box::new( + api_error( + StatusCode::INTERNAL_SERVER_ERROR, + "could not resolve recovery audit scope", + ) + .into_response(), + )) + } + } +} + +pub(super) async fn approve( + State(state): State, + Extension(AuthUser(actor)): Extension, + Path(sandbox): Path, + Json(request): Json, +) -> Response { + if let Err(error) = require_managing_org_admin(&state.pool, &actor).await { + return error.into_response(); + } + if !state.feature_flags.bool_value( + pplx_space_feature_flags::BACKUP_RECOVERY_GRANTS_ENABLED, + "cluster", + ) { + return api_error(StatusCode::SERVICE_UNAVAILABLE,"backup recovery approval is disabled until all backup services support grant enforcement").into_response(); + } + let captured_at = match timestamp(&request.captured_at) { + Ok(value) => value, + Err(message) => return api_error(StatusCode::BAD_REQUEST, message).into_response(), + }; + let expires_at = match timestamp(&request.expires_at) { + Ok(value) => value, + Err(message) => return api_error(StatusCode::BAD_REQUEST, message).into_response(), + }; + let mut tx = match state.pool.begin().await { + Ok(tx) => tx, + Err(error) => return store_error(error.into()), + }; + let org_id = match audit_org(&mut tx, sandbox).await { + Ok(org) => org, + Err(response) => return *response, + }; + let id = RecoveryGrantId::new(); + let result = recovery::approve( + &mut tx, + recovery::ApproveRecovery { + id, + sandbox_id: sandbox.into(), + artifact_id: request.local_artifact_id, + generation: request.execution_generation, + uploader_node_id: request.uploader_node_id, + uploader_node_uid: &request.uploader_node_uid, + captured_at, + capture_evidence: &request.capture_evidence, + expires_at, + actor: actor.id, + reason: &request.reason, + manifest: &request.manifest, + }, + ) + .await; + if let Err(error) = result { + return store_error(error); + } + if let Err(error) = tx.commit().await { + return store_error(error.into()); + } + audit::record_sandbox(AdminSandboxAuditEntry { + actor: &actor, + action: "backup-recovery-approve", + sandbox_id: sandbox, + org_id, + operation_id: None, + success: true, + error: None, + }); + Json(serde_json::json!({"grant_id":id,"status":"approved"})).into_response() +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(super) struct Revocation { + reason: String, +} +pub(super) async fn revoke( + State(state): State, + Extension(AuthUser(actor)): Extension, + Path((sandbox, id)): Path<(PublicSandboxId, RecoveryGrantId)>, + Json(request): Json, +) -> Response { + if let Err(error) = require_managing_org_admin(&state.pool, &actor).await { + return error.into_response(); + } + let mut tx = match state.pool.begin().await { + Ok(tx) => tx, + Err(error) => return store_error(error.into()), + }; + let org_id = match audit_org(&mut tx, sandbox).await { + Ok(org) => org, + Err(response) => return *response, + }; + match recovery::revoke(&mut tx, sandbox.into(), id, actor.id, &request.reason).await { + Ok(true) => {} + Ok(false) => { + return api_error( + StatusCode::CONFLICT, + "approval is absent, revoked, or consumed", + ) + .into_response(); + } + Err(error) => return store_error(error), + } + if let Err(error) = tx.commit().await { + return store_error(error.into()); + } + audit::record_sandbox(AdminSandboxAuditEntry { + actor: &actor, + action: "backup-recovery-revoke", + sandbox_id: sandbox, + org_id, + operation_id: None, + success: true, + error: None, + }); + Json(serde_json::json!({"grant_id":id,"status":"revoked"})).into_response() +} + +pub(super) async fn list( + State(state): State, + Extension(AuthUser(actor)): Extension, + Path(sandbox): Path, +) -> Response { + if let Err(error) = require_managing_org_admin(&state.pool, &actor).await { + return error.into_response(); + } + match recovery::list_approvals(&state.pool, sandbox.into()).await { + Ok(grants) => Json(serde_json::json!({"grants":grants})).into_response(), + Err(error) => store_error(error.into()), + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/libs/api/src/routes/admin/backup_recovery/tests.rs b/crates/libs/api/src/routes/admin/backup_recovery/tests.rs new file mode 100644 index 0000000000..54cf93913f --- /dev/null +++ b/crates/libs/api/src/routes/admin/backup_recovery/tests.rs @@ -0,0 +1,108 @@ +use super::*; +use crate::{models::User, user_role::UserRole}; +use pplx_space_common_core::{ + sql, + types::{Arch, ExecutionMetadata, SandboxId, SnapshotExecutionMetadata}, +}; + +#[tokio::test] +async fn recovery_admin_authority_and_rollout_gate() -> pplx_space_test_env::TestResult { + if !pplx_space_test_env::enter_test_process(&[( + "SPACE_K8S_CONTROL_PLANE", + Some(std::ffi::OsStr::new("false")), + )]) { + return Ok(()); + } + let Some(db) = pplx_space_db_test_fixtures::try_pg_test_db().await? else { + return Ok(()); + }; + let mut state = AppState::from_pool(db.pool.clone()).await?; + state.feature_flags = pplx_space_feature_flags::FeatureFlagClient::disabled(); + let admin = sqlx::query_as::<_, User>(sql!("SELECT u.* FROM users u JOIN clusters c ON c.managing_org_id=u.org_id WHERE c.singleton AND u.role='organization_admin' ORDER BY u.id LIMIT 1")) + .fetch_one(&db.pool).await?; + let sandbox = PublicSandboxId::from(SandboxId::new()); + for (role, enabled, expected) in [ + (UserRole::Member, true, StatusCode::FORBIDDEN), + ( + UserRole::OrganizationAdmin, + false, + StatusCode::SERVICE_UNAVAILABLE, + ), + (UserRole::OrganizationAdmin, true, StatusCode::NOT_FOUND), + ] { + let mut actor = admin.clone(); + actor.role = role; + state.feature_flags = pplx_space_feature_flags::FeatureFlagClient::fixed_bools([( + pplx_space_feature_flags::BACKUP_RECOVERY_GRANTS_ENABLED.key(), + enabled, + )]); + let approval = Approval { + local_artifact_id: LocalArtifactId::new(), + execution_generation: ExecutionGeneration::new(1)?, + uploader_node_id: NodeId::new(), + uploader_node_uid: "synthetic-node".parse()?, + captured_at: "2026-01-01T00:00:00Z".into(), + expires_at: "2026-01-01T01:00:00Z".into(), + capture_evidence: recovery::CaptureTimeEvidence::RecordedCapture { + reference: "synthetic capture".into(), + }, + reason: "synthetic approval".into(), + manifest: RecoveryManifest { + validation: recovery::SealedValidation { + sealed_full_suspend: true, + evidence_reference: "synthetic validation".into(), + }, + execution: SnapshotExecutionMetadata::new(ExecutionMetadata::initial_firecracker( + Arch::Aarch64, + vec![], + ))?, + metadata_binding_digest: format!("sha256:{}", "11".repeat(32)).parse()?, + vcpu: 2, + memory_mib: 1024, + template_id: "synthetic-template".parse()?, + content_hash: None, + artifacts: vec![], + }, + }; + let response = approve( + State(state.clone()), + Extension(AuthUser(actor.clone())), + Path(sandbox), + Json(approval), + ) + .await; + assert_eq!(response.status(), expected); + let response = list( + State(state.clone()), + Extension(AuthUser(actor.clone())), + Path(sandbox), + ) + .await; + assert_eq!( + response.status(), + if role == UserRole::Member { + StatusCode::FORBIDDEN + } else { + StatusCode::OK + } + ); + let response = revoke( + State(state.clone()), + Extension(AuthUser(actor)), + Path((sandbox, RecoveryGrantId::new())), + Json(Revocation { + reason: "synthetic revocation".into(), + }), + ) + .await; + assert_eq!( + response.status(), + if role == UserRole::Member { + StatusCode::FORBIDDEN + } else { + StatusCode::NOT_FOUND + } + ); + } + Ok(()) +} diff --git a/crates/libs/api/src/routes/admin/docs.rs b/crates/libs/api/src/routes/admin/docs.rs index 51965eb79c..4684374e3d 100644 --- a/crates/libs/api/src/routes/admin/docs.rs +++ b/crates/libs/api/src/routes/admin/docs.rs @@ -364,9 +364,187 @@ fn admin_openapi_value() -> Value { insert_admin_sandbox_create_paths(paths); insert_identity_profile_paths(paths); insert_audit_event_paths(paths); + insert_backup_recovery_paths(paths); spec } +fn insert_backup_recovery_paths(paths: &mut Map) { + paths.insert("/sandboxes/{sandbox_id}/backup-recovery-grants".to_owned(), json!({ + "get": { + "parameters": [ + { + "name": "sandbox_id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + } + ], + "summary": "List the newest 100 recovery approvals and their outcomes", + "responses": { + "200": { + "description": "Recovery approvals" + } + } + }, + "post": { + "parameters": [ + { + "name": "sandbox_id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + } + ], + "summary": "Approve one validated retained full suspend artifact", + "description": "Requires managing-organization admin authentication and the recovery rollout flag. Records exact logical file proofs, stable metadata digest, capture time, execution generation, and current uploader incarnation. Expires within 24 hours. Does not enqueue, seal, clear quarantine, or wake.", + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "local_artifact_id", + "execution_generation", + "uploader_node_id", + "uploader_node_uid", + "captured_at", + "capture_evidence", + "expires_at", + "reason", + "manifest" + ], + "properties": { + "local_artifact_id": { + "type": "string", + "format": "uuid" + }, + "execution_generation": { + "type": "integer", + "minimum": 1 + }, + "uploader_node_id": { + "type": "string", + "format": "uuid" + }, + "uploader_node_uid": { + "type": "string" + }, + "captured_at": { + "type": "string", + "format": "date-time" + }, + "capture_evidence": { + "description": "Capture time must fall within the approved execution lifetime.", + "oneOf": [ + { + "type": "object", "additionalProperties": false, + "required": ["kind", "reference"], + "properties": { + "kind": { "type": "string", "enum": ["recorded_capture"] }, + "reference": { "type": "string", "minLength": 1, "maxLength": 2048 } + } + }, + { + "type": "object", "additionalProperties": false, + "required": ["kind", "explanation"], + "properties": { + "kind": { "type": "string", "enum": ["operator_attests_no_newer_state"] }, + "explanation": { "type": "string", "minLength": 1, "maxLength": 2048 } + } + } + ] + }, + "expires_at": { + "type": "string", + "format": "date-time" + }, + "reason": { + "type": "string", + "maxLength": 2048 + }, + "manifest": { + "type": "object", + "description": "Requires validation {sealed_full_suspend: true, evidence_reference}, captured execution, vcpu, memory_mib, template_id, optional content_hash, metadata_binding_digest, and exact logical VM artifacts (artifact, digest, size_bytes). Generated btrfs-send sources are not supported." + } + } + } + } + } + }, + "responses": { + "200": { + "description": "Approval identity" + }, + "409": { + "description": "State or approval conflict" + }, + "503": { + "description": "Recovery issuance disabled" + } + } + } +})); + paths.insert( + "/sandboxes/{sandbox_id}/backup-recovery-grants/{grant_id}/revoke".to_owned(), + json!({ + "post": { + "summary": "Revoke an unconsumed recovery approval", + "parameters": [ + { + "name": "sandbox_id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + }, + { + "name": "grant_id", + "in": "path", + "required": true, + "schema": { + "type": "string", + "format": "uuid" + } + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "reason" + ], + "properties": { + "reason": { + "type": "string", + "maxLength": 2048 + } + } + } + } + } + }, + "responses": { + "200": { + "description": "Revoked" + }, + "409": { + "description": "Absent, already revoked, or consumed" + } + } + } + }), + ); +} + fn insert_api_key_paths(paths: &mut Map) { paths.insert( "/orgs/{org_id}/api-keys".to_string(), diff --git a/crates/libs/api/src/routes/admin/docs/snapshots/pplx_space_api__routes__admin__docs__tests__admin_openapi_stub_is_admin_only.snap b/crates/libs/api/src/routes/admin/docs/snapshots/pplx_space_api__routes__admin__docs__tests__admin_openapi_stub_is_admin_only.snap index 72bdcae016..2f07742895 100644 --- a/crates/libs/api/src/routes/admin/docs/snapshots/pplx_space_api__routes__admin__docs__tests__admin_openapi_stub_is_admin_only.snap +++ b/crates/libs/api/src/routes/admin/docs/snapshots/pplx_space_api__routes__admin__docs__tests__admin_openapi_stub_is_admin_only.snap @@ -1,5 +1,6 @@ --- source: crates/libs/api/src/routes/admin/docs/tests.rs +assertion_line: 34 expression: paths_by_method --- { @@ -97,6 +98,13 @@ expression: paths_by_method "/sandboxes/{sandbox_id}": [ "patch" ], + "/sandboxes/{sandbox_id}/backup-recovery-grants": [ + "get", + "post" + ], + "/sandboxes/{sandbox_id}/backup-recovery-grants/{grant_id}/revoke": [ + "post" + ], "/sandboxes/{sandbox_id}:force-{action}": [ "post" ], diff --git a/crates/libs/api/src/routes/admin/mod.rs b/crates/libs/api/src/routes/admin/mod.rs index 409c3abc83..33dc14b86f 100644 --- a/crates/libs/api/src/routes/admin/mod.rs +++ b/crates/libs/api/src/routes/admin/mod.rs @@ -36,6 +36,7 @@ use crate::{ mod api_key_store; mod api_keys; pub mod audit; +mod backup_recovery; mod docs; mod identity; mod node_group_store; @@ -143,6 +144,14 @@ pub fn v1_router(state: AppState) -> Router { "/nodes/{node_id}/templates/{template_name}", axum::routing::delete(node_maintenance::delete_node_template), ) + .route( + "/sandboxes/{sandbox_id}/backup-recovery-grants", + get(backup_recovery::list).post(backup_recovery::approve), + ) + .route( + "/sandboxes/{sandbox_id}/backup-recovery-grants/{grant_id}/revoke", + axum::routing::post(backup_recovery::revoke), + ) .route("/operations", get(operations::list_admin_operations)) .route( "/operations/{operation_id}", diff --git a/crates/libs/feature-flags/src/lib.rs b/crates/libs/feature-flags/src/lib.rs index f0a4a83859..81ca34976e 100644 --- a/crates/libs/feature-flags/src/lib.rs +++ b/crates/libs/feature-flags/src/lib.rs @@ -171,6 +171,11 @@ pub use provider::{ AttributeValue, Attributes, FeatureFlagProvider, ProviderState, UnavailableProvider, Unresolved, }; +/// Issuance is off until every backup reader enforces recovery provenance. +/// Turning it off prevents new approvals; existing approvals remain enforced. +pub const BACKUP_RECOVERY_GRANTS_ENABLED: BoolFlag = + BoolFlag::new("space-backup-recovery-grants-enabled", false); + /// Boolean flag acting as a dynamic cluster-wide kill-switch for the /// Redis-backed sandbox activation rate limiter. Layered under the static /// `SPACE_CLUSTER_ACTIVATION_ADMISSION_ENABLED` env master by [`policy`]. diff --git a/crates/libs/internal/api/server/src/backup_ownership.rs b/crates/libs/internal/api/server/src/backup_ownership.rs index 28bb628910..af0425c0f5 100644 --- a/crates/libs/internal/api/server/src/backup_ownership.rs +++ b/crates/libs/internal/api/server/src/backup_ownership.rs @@ -33,7 +33,7 @@ pub(super) struct Caller { identity: Option>, } impl Caller { - fn verified(&self) -> Result<&KubernetesPodIdentity, Status> { + pub(super) fn verified(&self) -> Result<&KubernetesPodIdentity, Status> { self.identity .as_ref() .ok_or_else(|| { @@ -233,3 +233,10 @@ impl ClusterApiService { Ok(()) } } + +pub(super) fn recovery_error(error: store::recovery::RecoveryError) -> Status { + match error { + store::recovery::RecoveryError::Refused(message) => Status::failed_precondition(message), + _ => Status::internal(format!("backup recovery: {error}")), + } +} diff --git a/crates/libs/internal/api/server/src/lib.rs b/crates/libs/internal/api/server/src/lib.rs index 561a114f84..1095ffbdab 100644 --- a/crates/libs/internal/api/server/src/lib.rs +++ b/crates/libs/internal/api/server/src/lib.rs @@ -761,6 +761,31 @@ impl ClusterApiService { self.authorize_backup_incarnation(&mut tx, &attempt, caller) .await?; + let recovery_to_consume = + if target_state.is_ready() && attempt.recovery_grant_id().is_some() { + let grant = snapshot_store::recovery::for_attempt(&mut tx, snapshot_id) + .await + .map_err(backup_ownership::recovery_error)? + .ok_or_else(|| Status::failed_precondition("recovery approval missing"))?; + if snapshot_store::recovery::validate_ready( + &mut tx, + &grant, + &attempt, + artifact_digests, + upload_evidence, + bytes_uploaded, + ) + .await + .map_err(backup_ownership::recovery_error)? + { + Some(grant) + } else { + None + } + } else { + None + }; + // `RETURNING` the sandbox so a READY transition can wake the // cohort-restore listener without a second lookup, and the node and // artifact identity so it can clear exactly this artifact's row from @@ -848,6 +873,11 @@ impl ClusterApiService { })?; } + if let Some(grant) = recovery_to_consume { + snapshot_store::recovery::consume(&mut tx, &grant, snapshot_id) + .await + .map_err(backup_ownership::recovery_error)?; + } let suspend_projection = if target_state.is_ready() { project_ready_suspend_snapshot(&mut tx, snapshot_id).await? } else { @@ -1069,6 +1099,67 @@ impl ClusterApiService { ) .await .map_err(backup_ownership::database_error)?; + let grant = match snapshot_store::recovery::for_attempt(&mut tx, snapshot_id) + .await + .map_err(backup_ownership::recovery_error)? + { + Some(grant) => Some(grant), + None => snapshot_store::recovery::for_artifact(&mut tx, sandbox_id, local_artifact_id) + .await + .map_err(backup_ownership::recovery_error)?, + }; + if let Some(grant) = grant { + if execution_generation.map(ExecutionGeneration::get) + != Some(grant.execution_generation) + { + return Err(Status::failed_precondition( + "recovery request does not name the approved execution", + )); + } + let identity = caller.verified()?; + self.authorize_backup_source_node( + &mut *tx, + Some(grant.uploader_node_id), + caller.remote_addr, + "recovery uploader", + ) + .await?; + let persisted = + snapshot_store::recovery::register(&mut tx, &grant, snapshot, identity.node_uid()) + .await + .map_err(backup_ownership::recovery_error)?; + self.authorize_backup_incarnation(&mut tx, &persisted, &caller) + .await?; + if req.fence_verdict_only { + tx.rollback() + .await + .map_err(backup_ownership::database_error)?; + } else { + tx.commit() + .await + .map_err(backup_ownership::database_error)?; + } + return Ok(Response::new(BeginArtifactBackupResponse { + local_artifact_id: local_artifact_id.to_string(), + backup_attempt_id: backup_attempt_id.to_string(), + fence_verdict_only: req.fence_verdict_only, + })); + } + // An explicit recovery approval reserves subsequent registrations of + // these bytes for its approved uploader. Historical ordinary attempts + // must neither block that approval nor bypass it after consumption. + if snapshot_store::backup_authority::has_recovery_provenance( + &mut tx, + sandbox_id, + local_artifact_id, + ) + .await + .map_err(backup_ownership::database_error)? + { + return Err(Status::failed_precondition( + "recovery artifact requires a new valid approval", + )); + } snapshot = snapshot.with_source_node_id( prior .as_ref() diff --git a/crates/libs/internal/api/server/src/tests/backup_identity.rs b/crates/libs/internal/api/server/src/tests/backup_identity.rs index deeeac5c1e..632fb6509b 100644 --- a/crates/libs/internal/api/server/src/tests/backup_identity.rs +++ b/crates/libs/internal/api/server/src/tests/backup_identity.rs @@ -277,8 +277,324 @@ async fn verified_backup_rpcs_preserve_retry_authority_and_reject_replacement( client .finalize_artifact_backup(authenticated(finalize(), &token)) .await?; + exercise_recovery(&mut client, &pool, &token, node).await?; drop(client); drop(rpc); drop(api); Ok(()) } + +async fn exercise_recovery( + client: &mut ClusterApiClient, + pool: &sqlx::PgPool, + token: &str, + node: NodeId, +) -> TestResult { + use pplx_space_common_core::types::{K8sNodeUid, SnapshotArtifact as File}; + use snapshot_store::recovery::{ + self, ApproveRecovery, CaptureTimeEvidence, RecoveryGrantId, RecoveryManifest, + SealedValidation, + }; + let actor = UserId::from_uuid(SEEDED_USER_ID); + let uid: K8sNodeUid = "node-original".parse()?; + for scenario in [ + "ready", + "prior-registration", + "expired", + "revoked", + "wrong-proof", + "wrong-object", + "changed-state", + "newer-materialized", + "abandoned-newer", + "race-revoke", + ] { + let sandbox = seed_unplaced_sandbox(pool).await?; + let artifact = LocalArtifactId::new(); + sqlx::query(sql!("UPDATE sandboxes SET lifecycle='quarantined',current_local_artifact_id=$2,current_local_artifact_kind='full',current_local_artifact_origin='suspend' WHERE id=$1")).bind(sandbox).bind(artifact).execute(pool).await?; + sqlx::query(sql!("INSERT INTO sandbox_k8s_executions(sandbox_id,generation,pod_name,pod_namespace,state,active,created_at,terminated_at) + VALUES($1,1,'retained-' || $1::text,'space','retired',false,NOW()-INTERVAL '2 hours',NOW()-INTERVAL '1 hour')")).bind(sandbox).execute(pool).await?; + if scenario == "abandoned-newer" { + sqlx::query(sql!("INSERT INTO sandbox_k8s_executions(sandbox_id,generation,pod_name,pod_namespace,state,active) VALUES($1,2,'abandoned-pod','space','failed',false)")).bind(sandbox).execute(pool).await?; + } + let manifest = RecoveryManifest { + validation: SealedValidation { + sealed_full_suspend: true, + evidence_reference: "synthetic validation of sealed fixture".into(), + }, + execution: snapshot_execution_metadata(Some(&test_snapshot_execution()))?, + metadata_binding_digest: format!("sha256:{}", "33".repeat(32)).parse()?, + vcpu: 2, + memory_mib: 1024, + template_id: "synthetic-template".parse()?, + content_hash: None, + artifacts: [File::Rootfs, File::Memory, File::VmState] + .into_iter() + .map(|artifact| snapshot_store::SnapshotArtifactDigest { + artifact, + digest: format!("sha256:{}", "22".repeat(32)).parse().unwrap(), + size_bytes: 32, + }) + .collect(), + }; + let id = RecoveryGrantId::new(); + let capture_evidence = CaptureTimeEvidence::RecordedCapture { + reference: "synthetic durable capture record".into(), + }; + let (captured_at, expires_at): (OffsetDateTime, OffsetDateTime) = sqlx::query_as(sql!( + "SELECT NOW()-INTERVAL '90 minutes',NOW()+INTERVAL '1 hour'" + )) + .fetch_one(pool) + .await?; + if scenario == "prior-registration" { + // A failed upload from the old incarnation must not prevent an + // explicit approval for the currently authenticated uploader. + sqlx::query(sql!("INSERT INTO snapshots(id,sandbox_id,type,state,timestamp,source_node_id,source_node_uid,local_artifact_id,artifact_kind,artifact_origin,generation) + VALUES($1,$2,'full','failed',$3,$4,'prior-incarnation',$5,'full','suspend',1)")) + .bind(SnapshotId::new()).bind(sandbox).bind(captured_at).bind(node).bind(artifact).execute(pool).await?; + } + let mut tx = pool.begin().await?; + recovery::approve( + &mut tx, + ApproveRecovery { + id, + sandbox_id: sandbox, + artifact_id: artifact, + generation: ExecutionGeneration::new(1)?, + uploader_node_id: node, + uploader_node_uid: &uid, + captured_at, + capture_evidence: &capture_evidence, + expires_at, + actor, + reason: "validate retained fixture", + manifest: &manifest, + }, + ) + .await?; + tx.commit().await?; + if scenario == "ready" { + // The SQL constraints protect audit consistency even from a caller + // that bypasses the typed approval route. + for mutation in [ + "UPDATE backup_recovery_grants SET expires_at=approved_at+INTERVAL '25 hours' WHERE id=$1", + "UPDATE backup_recovery_grants SET revoked_at=NOW() WHERE id=$1", + "UPDATE backup_recovery_grants SET consumed_attempt_id=space_uuid_v7() WHERE id=$1", + "UPDATE backup_recovery_grants SET active_attempt_id=id,consumed_attempt_id=id,revoked_at=NOW(),revoked_by=approved_by,revoke_reason='conflict' WHERE id=$1", + ] { + let mut constraint_tx = pool.begin().await?; + let error = sqlx::query(sqlx::AssertSqlSafe(sql!(mutation))) + .bind(id) + .execute(&mut *constraint_tx) + .await + .unwrap_err(); + assert_eq!( + error + .as_database_error() + .and_then(|error| error.code()) + .as_deref(), + Some("23514") + ); + constraint_tx.rollback().await?; + } + } + let attempt = SnapshotId::new(); + let begin = |attempt: SnapshotId| BeginArtifactBackupRequest { + sandbox_id: sandbox.to_string(), + local_artifact_id: artifact.to_string(), + backup_attempt_id: attempt.to_string(), + storage_uri: test_storage_uri(sandbox, attempt), + artifact_kind: pplx_space_sbx_mgr_client::proto::ArtifactKind::Full as i32, + artifact_origin: pplx_space_sbx_mgr_client::proto::ArtifactOrigin::Suspend as i32, + execution: Some(test_snapshot_execution()), + execution_generation: Some(1), + ..Default::default() + }; + client + .begin_artifact_backup(authenticated(begin(attempt), token)) + .await?; + assert_eq!( + client + .begin_artifact_backup(authenticated(begin(SnapshotId::new()), token)) + .await + .unwrap_err() + .code(), + Code::FailedPrecondition, + "one active attempt: {scenario}" + ); + let captured: bool = sqlx::query_scalar(sql!( + "SELECT timestamp=$2 AND last_attempt_at>timestamp FROM snapshots WHERE id=$1" + )) + .bind(attempt) + .bind(captured_at) + .fetch_one(pool) + .await?; + assert!(captured, "grant must preserve capture time"); + let finalize = |attempt: SnapshotId| { + let uri = pplx_space_common_core::types::StorageUri::parse(&test_storage_uri( + sandbox, attempt, + )) + .unwrap(); + let mut artifact_digests: Vec<_> = manifest + .artifacts + .iter() + .map(|file| crate::proto::SnapshotArtifactDigest { + artifact: file.artifact.file_name().into(), + digest: file.digest.to_string(), + size_bytes: file.size_bytes, + upload_evidence: Some(crate::proto::SnapshotUploadEvidence { + object_key: format!("{}/{}", uri.key_prefix, file.artifact.file_name()), + wire_size_bytes: 16, + etag: Some("synthetic-object".into()), + version_id: None, + metadata_binding_digest: None, + }), + }) + .collect(); + artifact_digests.push(crate::proto::SnapshotArtifactDigest { + artifact: "metadata.json".into(), + digest: format!("sha256:{}", "44".repeat(32)), + size_bytes: 5, + upload_evidence: Some(crate::proto::SnapshotUploadEvidence { + object_key: format!("{}/metadata.json", uri.key_prefix), + wire_size_bytes: 5, + etag: Some("synthetic-metadata".into()), + version_id: None, + metadata_binding_digest: Some(manifest.metadata_binding_digest.to_string()), + }), + }); + FinalizeArtifactBackupRequest { + backup_attempt_id: attempt.to_string(), + target_state: SuspendSnapshotTerminalState::Ready as i32, + bytes_uploaded: 48, + artifact_digests, + failure_reason: None, + } + }; + match scenario { + "expired" => { + sqlx::query(sql!("UPDATE backup_recovery_grants SET approved_at=NOW()-INTERVAL '1 hour',expires_at=NOW()-INTERVAL '1 minute' WHERE id=$1")).bind(id).execute(pool).await?; + } + "revoked" => { + let mut tx = pool.begin().await?; + assert!(recovery::revoke(&mut tx, sandbox, id, actor, "synthetic revoke").await?); + tx.commit().await?; + } + "changed-state" => { + sqlx::query(sql!("UPDATE sandboxes SET lifecycle='failed' WHERE id=$1")) + .bind(sandbox) + .execute(pool) + .await?; + } + "newer-materialized" => { + sqlx::query(sql!("INSERT INTO sandbox_k8s_executions(sandbox_id,generation,pod_name,pod_namespace,state,active,pod_materialized_at) VALUES($1,2,'newer-pod','space','retired',false,NOW())")).bind(sandbox).execute(pool).await?; + } + _ => (), + } + let mut request = finalize(attempt); + if scenario == "wrong-proof" { + request.artifact_digests[0].digest = format!("sha256:{}", "55".repeat(32)); + } + if scenario == "wrong-object" { + request.artifact_digests[0] + .upload_evidence + .as_mut() + .unwrap() + .object_key = "other-attempt/rootfs".into(); + } + if scenario == "race-revoke" { + let mut gate = pool.begin().await?; + sqlx::query(sql!("SELECT id FROM sandboxes WHERE id=$1 FOR UPDATE")) + .bind(sandbox) + .execute(&mut *gate) + .await?; + let mut writer = client.clone(); + let token = token.to_owned(); + let ready = tokio::spawn(async move { + writer + .finalize_artifact_backup(authenticated(request, &token)) + .await + }); + let revoke_pool = pool.clone(); + let revoke = tokio::spawn(async move { + let mut tx = revoke_pool.begin().await?; + let revoked = + recovery::revoke(&mut tx, sandbox, id, actor, "concurrent revoke").await?; + tx.commit().await?; + Ok::<_, recovery::RecoveryError>(revoked) + }); + gate.commit().await?; + let ready = ready.await?; + let revoked = revoke.await??; + assert_ne!( + ready.is_ok(), + revoked, + "exactly one Ready/revoke transition may win" + ); + let outcome:(bool,bool,bool)=sqlx::query_as(sql!("SELECT s.state='ready',g.consumed_attempt_id IS NOT DISTINCT FROM s.id,g.revoked_at IS NOT NULL FROM snapshots s JOIN backup_recovery_grants g ON s.recovery_grant_id=g.id WHERE s.id=$1")).bind(attempt).fetch_one(pool).await?; + assert_eq!(outcome.0, outcome.1); + assert_ne!(outcome.0, outcome.2); + continue; + } + let result = client + .finalize_artifact_backup(authenticated(request, token)) + .await; + if matches!(scenario, "ready" | "abandoned-newer" | "prior-registration") { + result?; + client + .begin_artifact_backup(authenticated(begin(attempt), token)) + .await?; + let mut altered = begin(attempt); + altered.storage_uri = altered.storage_uri.replace("test-bucket", "other-bucket"); + assert_eq!( + client + .begin_artifact_backup(authenticated(altered, token)) + .await + .unwrap_err() + .code(), + Code::FailedPrecondition + ); + client + .finalize_artifact_backup(authenticated(finalize(attempt), token)) + .await?; + assert_eq!( + client + .begin_artifact_backup(authenticated(begin(SnapshotId::new()), token)) + .await + .unwrap_err() + .code(), + Code::FailedPrecondition, + "consumed grant cannot confer ordinary retry authority" + ); + let row:(String,bool)=sqlx::query_as(sql!("SELECT s.lifecycle,g.consumed_attempt_id=$2 FROM sandboxes s JOIN backup_recovery_grants g ON g.sandbox_id=s.id WHERE g.id=$1")).bind(id).bind(attempt).fetch_one(pool).await?; + assert_eq!(row, ("quarantined".into(), true)); + } else { + assert_eq!( + result.unwrap_err().code(), + Code::FailedPrecondition, + "{scenario}" + ); + let state: String = sqlx::query_scalar(sql!("SELECT state FROM snapshots WHERE id=$1")) + .bind(attempt) + .fetch_one(pool) + .await?; + assert_eq!(state, "creating", "failed validation must roll back Ready"); + let mut failed = finalize(attempt); + failed.target_state = SuspendSnapshotTerminalState::Failed as i32; + failed.artifact_digests.clear(); + client + .finalize_artifact_backup(authenticated(failed, token)) + .await?; + if scenario == "wrong-proof" { + let retry = SnapshotId::new(); + client + .begin_artifact_backup(authenticated(begin(retry), token)) + .await?; + client + .finalize_artifact_backup(authenticated(finalize(retry), token)) + .await?; + } + } + } + Ok(()) +} diff --git a/crates/libs/snapshot/store/src/backup_authority.rs b/crates/libs/snapshot/store/src/backup_authority.rs index 5d7a37fd7f..a0d5b23445 100644 --- a/crates/libs/snapshot/store/src/backup_authority.rs +++ b/crates/libs/snapshot/store/src/backup_authority.rs @@ -31,7 +31,7 @@ pub async fn prior_registration( artifact: LocalArtifactId, ) -> Result, sqlx::Error> { sqlx::query_as(sql!("SELECT id,source_node_id,source_node_uid FROM snapshots - WHERE sandbox_id=$1 AND local_artifact_id=$2 AND source_node_id IS NOT NULL AND source_node_uid IS NOT NULL + WHERE recovery_grant_id IS NULL AND sandbox_id=$1 AND local_artifact_id=$2 AND source_node_id IS NOT NULL AND source_node_uid IS NOT NULL ORDER BY timestamp,id LIMIT 1")) .bind(sandbox).bind(artifact).fetch_optional(conn).await } @@ -133,3 +133,13 @@ pub async fn touch( .await?; Ok(()) } + +/// Grant-backed bytes never become ordinary retry authority after expiry/consumption. +pub async fn has_recovery_provenance( + conn: &mut sqlx::PgConnection, + sandbox: SandboxId, + artifact: LocalArtifactId, +) -> Result { + sqlx::query_scalar(sql!("SELECT EXISTS(SELECT 1 FROM snapshots WHERE sandbox_id=$1 AND local_artifact_id=$2 AND recovery_grant_id IS NOT NULL) OR EXISTS(SELECT 1 FROM backup_recovery_grants WHERE sandbox_id=$1 AND local_artifact_id=$2)")) + .bind(sandbox).bind(artifact).fetch_one(conn).await +} diff --git a/crates/libs/snapshot/store/src/lib.rs b/crates/libs/snapshot/store/src/lib.rs index c392d3ea18..4e1e47ba7c 100644 --- a/crates/libs/snapshot/store/src/lib.rs +++ b/crates/libs/snapshot/store/src/lib.rs @@ -20,6 +20,7 @@ use time::OffsetDateTime; use uuid::Uuid; pub mod backup_authority; +pub mod recovery; pub mod upload_evidence; /// Reading a snapshot catalog row failed. @@ -232,6 +233,7 @@ pub struct SnapshotRecord { timestamp: Timestamp, source_node_id: Option, source_node_uid: Option, + recovery_grant_id: Option, label: Option, retention_tier: Option, vcpu: Option, @@ -273,6 +275,10 @@ impl SnapshotRecord { self.source_node_id } + pub fn recovery_grant_id(&self) -> Option { + self.recovery_grant_id + } + pub fn source_node_uid(&self) -> Option<&K8sNodeUid> { self.source_node_uid.as_ref() } @@ -457,6 +463,7 @@ struct SnapshotRow { timestamp: OffsetDateTime, source_node_id: Option, source_node_uid: Option, + recovery_grant_id: Option, label: Option, retention_tier: Option, vcpu: Option, @@ -527,7 +534,7 @@ pub async fn delete_snapshot_catalog( /// Content digest of one artifact file in a snapshot attempt, as reported by /// the node that uploaded it. Restore compares the bytes it pulls against this /// instead of trusting the object store. -#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)] pub struct SnapshotArtifactDigest { pub artifact: SnapshotArtifactFile, pub digest: Digest, @@ -813,7 +820,7 @@ pub async fn artifact_registered_by_another_attempt( "SELECT EXISTS ( SELECT 1 FROM snapshots - WHERE sandbox_id = $1 + WHERE recovery_grant_id IS NULL AND sandbox_id = $1 AND local_artifact_id = $2 AND id <> $3 )" @@ -862,7 +869,7 @@ pub async fn get( ) -> Result, GetSnapshotError> { let row: Option = sqlx::query_as(sql!( "SELECT sn.id, sn.sandbox_id, sn.type AS snapshot_kind, sn.state, - sn.timestamp, sn.source_node_id, sn.source_node_uid, sn.label, sn.retention_tier, + sn.timestamp, sn.source_node_id, sn.source_node_uid, sn.recovery_grant_id, sn.label, sn.retention_tier, sn.vcpu, sn.memory_mib, sn.size_bytes, sn.template_id, sn.content_hash, sn.org_id, sn.failure_reason, sn.last_attempt_at, sn.local_artifact_id, sn.artifact_kind, @@ -957,6 +964,7 @@ impl TryFrom for SnapshotRecord { timestamp, source_node_id: row.source_node_id, source_node_uid: row.source_node_uid, + recovery_grant_id: row.recovery_grant_id, label: row.label, retention_tier: row.retention_tier, vcpu: row.vcpu, diff --git a/crates/libs/snapshot/store/src/recovery.rs b/crates/libs/snapshot/store/src/recovery.rs new file mode 100644 index 0000000000..afce5f1fe1 --- /dev/null +++ b/crates/libs/snapshot/store/src/recovery.rs @@ -0,0 +1,609 @@ +//! Explicit, bounded recovery authority. A grant proves an administrator's +//! approval of retained bytes, never their historical machine ownership. +use super::*; +use pplx_space_common_core::types::{SubstrateKind, UserId}; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Copy, Debug, Eq, PartialEq, sqlx::Type, Serialize, Deserialize)] +#[sqlx(transparent)] +#[serde(transparent)] +pub struct RecoveryGrantId(Uuid); +impl RecoveryGrantId { + pub fn new() -> Self { + Self(Uuid::now_v7()) + } +} +impl Default for RecoveryGrantId { + fn default() -> Self { + Self::new() + } +} +impl std::fmt::Display for RecoveryGrantId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.0.fmt(f) + } +} +impl FromStr for RecoveryGrantId { + type Err = uuid::Error; + fn from_str(value: &str) -> Result { + Uuid::parse_str(value).map(Self) + } +} + +/// Stable metadata proof excludes generated timestamps and measured sizes. +/// File proofs cover logical bytes read by volume-manager, not transport objects. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RecoveryManifest { + /// The administrator has inspected a sealed full suspend and validated its + /// files against this manifest. Issuance does not seal or repair a capture. + pub validation: SealedValidation, + pub execution: SnapshotExecutionMetadata, + pub metadata_binding_digest: Digest, + pub vcpu: i32, + pub memory_mib: i32, + pub template_id: TemplateName, + pub content_hash: Option, + pub artifacts: Vec, +} +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SealedValidation { + pub sealed_full_suspend: bool, + pub evidence_reference: String, +} + +impl RecoveryManifest { + pub fn validate(&self) -> Result<(), RecoveryError> { + if !self.validation.sealed_full_suspend + || self.validation.evidence_reference.trim().is_empty() + || self.validation.evidence_reference.len() > 2048 + { + return Err(RecoveryError::Refused( + "recovery requires an operator-validated, already sealed full suspend", + )); + } + use SnapshotArtifactFile::*; + let mut files: Vec<_> = self.artifacts.iter().map(|p| p.artifact).collect(); + files.sort(); + let valid = [ + vec![Rootfs, Memory, VmState], + vec![EncryptedSnapshotVolume, EncryptedRootfsCow], + ] + .into_iter() + .any(|mut set| { + set.sort(); + set == files + }); + if !valid + || self + .artifacts + .iter() + .any(|p| p.size_bytes == 0 || p.size_bytes > i64::MAX as u64) + { + return Err(RecoveryError::Refused( + "recovery requires an exact, nonempty full-artifact file manifest", + )); + } + if self.vcpu <= 0 || self.memory_mib <= 0 { + return Err(RecoveryError::Refused( + "recovery requires the captured VM resource shape", + )); + } + self.total_bytes()?; + Ok(()) + } + pub fn total_bytes(&self) -> Result { + self.artifacts + .iter() + .try_fold(0_i64, |sum, p| { + i64::try_from(p.size_bytes) + .ok() + .and_then(|n| sum.checked_add(n)) + }) + .ok_or(RecoveryError::Refused( + "recovery manifest size exceeds catalog range", + )) + } + pub fn matches(&self, proofs: &[SnapshotArtifactDigest], metadata: Option) -> bool { + let mut expected = self.artifacts.clone(); + expected.sort_by_key(|p| p.artifact); + let mut actual = proofs.to_vec(); + actual.sort_by_key(|p| p.artifact); + expected == actual && metadata == Some(self.metadata_binding_digest) + } +} + +#[derive(Debug, thiserror::Error)] +pub enum RecoveryError { + #[error(transparent)] + Database(#[from] sqlx::Error), + #[error(transparent)] + Insert(#[from] InsertSnapshotError), + #[error(transparent)] + Read(#[from] GetSnapshotError), + #[error("{0}")] + Refused(&'static str), +} + +/// The administrator attests capture freshness, not historical node ownership. +/// The database additionally rejects times outside the known execution lifetime. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum CaptureTimeEvidence { + RecordedCapture { reference: String }, + OperatorAttestsNoNewerState { explanation: String }, +} +impl CaptureTimeEvidence { + fn validate(&self) -> Result<(), RecoveryError> { + let detail = match self { + Self::RecordedCapture { reference } => reference, + Self::OperatorAttestsNoNewerState { explanation } => explanation, + }; + if detail.trim().is_empty() || detail.len() > 2048 { + return Err(RecoveryError::Refused( + "capture evidence must contain 1 to 2048 bytes of explanation or reference", + )); + } + Ok(()) + } +} + +pub struct ApproveRecovery<'a> { + pub id: RecoveryGrantId, + pub sandbox_id: SandboxId, + pub artifact_id: LocalArtifactId, + pub generation: ExecutionGeneration, + pub uploader_node_id: NodeId, + pub uploader_node_uid: &'a K8sNodeUid, + pub captured_at: OffsetDateTime, + pub capture_evidence: &'a CaptureTimeEvidence, + pub expires_at: OffsetDateTime, + pub actor: UserId, + pub reason: &'a str, + pub manifest: &'a RecoveryManifest, +} + +#[derive(Debug, FromRow)] +pub struct RecoveryGrant { + pub id: RecoveryGrantId, + pub sandbox_id: SandboxId, + pub local_artifact_id: LocalArtifactId, + pub execution_generation: i64, + pub captured_at: OffsetDateTime, + pub uploader_node_id: NodeId, + pub uploader_node_uid: K8sNodeUid, + pub manifest: Json, + pub expected_state: Json, + pub active_attempt_id: Option, + pub consumed_attempt_id: Option, + pub usable: bool, +} + +/// Caller must hold the sandbox row lock before acquiring grant or node locks. +/// The predicate is also checked at finalization, so approval is not a stale +/// authorization to restore across a successor execution. +async fn check_retained_state( + conn: &mut sqlx::PgConnection, + sandbox: SandboxId, + generation: i64, + artifact: LocalArtifactId, + expected_state: Option<&serde_json::Value>, +) -> Result { + let _sandbox: Option = + sqlx::query_scalar(sql!("SELECT id FROM sandboxes WHERE id=$1 FOR UPDATE")) + .bind(sandbox) + .fetch_optional(&mut *conn) + .await?; + let _executions:Vec = sqlx::query_scalar(sql!("SELECT generation FROM sandbox_k8s_executions WHERE sandbox_id=$1 AND generation>=$2 FOR SHARE")) + .bind(sandbox).bind(generation).fetch_all(&mut *conn).await?; + let row: Option<(Option,bool,serde_json::Value)> = sqlx::query_as(sql!( + "SELECT current_local_artifact_id, + active_node_id IS NULL AND lifecycle IN ('suspended','quarantined','stopped','failed') + AND EXISTS(SELECT 1 FROM sandbox_k8s_executions e WHERE e.sandbox_id=s.id AND e.generation=$2) + AND NOT EXISTS(SELECT 1 FROM sandbox_k8s_executions e WHERE e.sandbox_id=s.id AND + (e.active OR (e.generation>$2 AND (e.pod_materialized_at IS NOT NULL OR NULLIF(BTRIM(e.pod_uid),'') IS NOT NULL)) + OR (e.generation=$2 AND e.state='replaced'))), + jsonb_build_object('lifecycle',lifecycle,'placement',placement,'active_node_id',active_node_id, + 'target_node_id',target_node_id,'artifact_id',current_local_artifact_id, + 'artifact_kind',current_local_artifact_kind,'artifact_origin',current_local_artifact_origin, + 'placement_changed_at',placement_changed_at) + FROM sandboxes s WHERE id=$1 FOR UPDATE" + )).bind(sandbox).bind(generation).fetch_optional(&mut *conn).await?; + let (pointer, eligible, witness) = + row.ok_or(RecoveryError::Refused("recovery sandbox is absent"))?; + if !eligible + || pointer.is_some_and(|id| id != artifact) + || expected_state.is_some_and(|expected| expected != &witness) + { + return Err(RecoveryError::Refused( + "recovery state changed or its execution was superseded", + )); + } + Ok(witness) +} + +async fn check_uploader( + conn: &mut sqlx::PgConnection, + node: NodeId, + uid: &K8sNodeUid, +) -> Result<(), RecoveryError> { + let eligible: Option = sqlx::query_scalar(sql!( + "SELECT n.role='worker' AND n.managed_by=$3 AND n.state<>'deleted' AND k.k8s_node_uid=$2 + FROM nodes n JOIN node_k8s k ON k.node_id=n.id WHERE n.id=$1 FOR SHARE OF n,k" + )) + .bind(node) + .bind(uid) + .bind(SubstrateKind::Kubernetes) + .fetch_optional(conn) + .await?; + if eligible != Some(true) { + return Err(RecoveryError::Refused( + "recovery uploader is not the approved live node incarnation", + )); + } + Ok(()) +} + +/// Admin route owns authentication. Actor and exact approval are durable in the +/// same transaction; no worker-facing RPC can create or extend this authority. +pub async fn approve( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + request: ApproveRecovery<'_>, +) -> Result<(), RecoveryError> { + request.manifest.validate()?; + request.capture_evidence.validate()?; + if request.reason.trim().is_empty() || request.reason.len() > 2048 { + return Err(RecoveryError::Refused( + "an approval reason of at most 2048 bytes is required", + )); + } + let witness = check_retained_state( + tx, + request.sandbox_id, + request.generation.get(), + request.artifact_id, + None, + ) + .await?; + let within_execution: bool = sqlx::query_scalar(sql!( + "SELECT EXISTS(SELECT 1 FROM sandbox_k8s_executions WHERE sandbox_id=$1 AND generation=$2 + AND created_at <= $3 AND $3 <= COALESCE(terminated_at,NOW()))" + )) + .bind(request.sandbox_id) + .bind(request.generation.get()) + .bind(request.captured_at) + .fetch_one(&mut **tx) + .await?; + if !within_execution { + return Err(RecoveryError::Refused( + "capture time is outside the approved execution lifetime", + )); + } + // Serialize grant replacement through the sandbox, including expired grants + // whose accepted uploads may still be running. Only one approval is live. + let occupied: bool = sqlx::query_scalar(sql!("SELECT EXISTS(SELECT 1 FROM backup_recovery_grants WHERE sandbox_id=$1 AND local_artifact_id=$2 AND revoked_at IS NULL AND consumed_attempt_id IS NULL AND expires_at>clock_timestamp())")) + .bind(request.sandbox_id).bind(request.artifact_id).fetch_one(&mut **tx).await?; + if occupied { + return Err(RecoveryError::Refused( + "an unexpired recovery approval already exists; revoke it before replacing it", + )); + } + check_uploader(tx, request.uploader_node_id, request.uploader_node_uid).await?; + let inserted = sqlx::query(sql!( + "INSERT INTO backup_recovery_grants(id,sandbox_id,local_artifact_id,execution_generation,uploader_node_id,uploader_node_uid,captured_at,expires_at,approved_by,approval_reason,manifest,expected_state,capture_evidence) + SELECT $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13 + WHERE $7<=NOW() AND $8>NOW() AND $8<=NOW()+INTERVAL '24 hours'" + )).bind(request.id).bind(request.sandbox_id).bind(request.artifact_id).bind(request.generation.get()) + .bind(request.uploader_node_id).bind(request.uploader_node_uid).bind(request.captured_at).bind(request.expires_at) + .bind(request.actor).bind(request.reason.trim()).bind(Json(request.manifest)).bind(Json(witness)).bind(Json(request.capture_evidence)).execute(&mut **tx).await?; + if inserted.rows_affected() != 1 { + return Err(RecoveryError::Refused( + "capture must be in the past and approval must expire within 24 hours", + )); + } + Ok(()) +} + +pub async fn revoke( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + sandbox: SandboxId, + id: RecoveryGrantId, + actor: UserId, + reason: &str, +) -> Result { + if reason.trim().is_empty() || reason.len() > 2048 { + return Err(RecoveryError::Refused( + "a revocation reason of at most 2048 bytes is required", + )); + } + sqlx::query(sql!("SELECT id FROM sandboxes WHERE id=$1 FOR UPDATE")) + .bind(sandbox) + .fetch_optional(&mut **tx) + .await?; + let result = sqlx::query(sql!("UPDATE backup_recovery_grants SET revoked_at=NOW(),revoked_by=$3,revoke_reason=$4 WHERE id=$1 AND sandbox_id=$2 AND revoked_at IS NULL AND consumed_attempt_id IS NULL")) + .bind(id).bind(sandbox).bind(actor).bind(reason.trim()).execute(&mut **tx).await?; + Ok(result.rows_affected() == 1) +} + +pub async fn for_artifact( + conn: &mut sqlx::PgConnection, + sandbox: SandboxId, + artifact: LocalArtifactId, +) -> Result, RecoveryError> { + Ok(sqlx::query_as(sql!("SELECT id,sandbox_id,local_artifact_id,execution_generation,captured_at,uploader_node_id,uploader_node_uid,manifest,expected_state,active_attempt_id,consumed_attempt_id, + revoked_at IS NULL AND expires_at>clock_timestamp() AND consumed_attempt_id IS NULL AS usable + FROM backup_recovery_grants WHERE sandbox_id=$1 AND local_artifact_id=$2 AND revoked_at IS NULL AND consumed_attempt_id IS NULL AND expires_at>clock_timestamp() ORDER BY approved_at DESC,id DESC LIMIT 1 FOR UPDATE")) + .bind(sandbox).bind(artifact).fetch_optional(conn).await?) +} + +pub async fn for_attempt( + conn: &mut sqlx::PgConnection, + attempt: SnapshotId, +) -> Result, RecoveryError> { + Ok(sqlx::query_as(sql!("SELECT g.id,g.sandbox_id,g.local_artifact_id,g.execution_generation,g.captured_at,g.uploader_node_id,g.uploader_node_uid,g.manifest,g.expected_state,g.active_attempt_id,g.consumed_attempt_id, + g.revoked_at IS NULL AND g.expires_at>clock_timestamp() AND g.consumed_attempt_id IS NULL AS usable + FROM backup_recovery_grants g JOIN snapshots s ON s.recovery_grant_id=g.id WHERE s.id=$1 FOR UPDATE OF g")) + .bind(attempt).fetch_optional(conn).await?) +} + +impl RecoveryGrant { + pub async fn validate_live( + &self, + conn: &mut sqlx::PgConnection, + caller_uid: &K8sNodeUid, + ) -> Result<(), RecoveryError> { + if !self.usable { + return Err(RecoveryError::Refused( + "recovery approval is expired, revoked, or consumed", + )); + } + if caller_uid != &self.uploader_node_uid { + return Err(RecoveryError::Refused( + "caller is not the approved recovery uploader", + )); + } + check_retained_state( + conn, + self.sandbox_id, + self.execution_generation, + self.local_artifact_id, + Some(&self.expected_state.0), + ) + .await?; + let newer: bool = sqlx::query_scalar(sql!("SELECT EXISTS(SELECT 1 FROM snapshots WHERE sandbox_id=$1 AND local_artifact_id IS DISTINCT FROM $2 AND timestamp>$3 AND state<>'deleting')")) + .bind(self.sandbox_id).bind(self.local_artifact_id).bind(self.captured_at).fetch_one(&mut *conn).await?; + if newer { + return Err(RecoveryError::Refused( + "newer captured state supersedes this recovery", + )); + } + check_uploader(conn, self.uploader_node_id, caller_uid).await + } +} + +/// Only this entry point can attach grant provenance to a fresh attempt. +/// Ordinary retry admission and store retry validation exclude all such rows. +pub async fn register( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + grant: &RecoveryGrant, + mut snapshot: NewSnapshot, + caller_uid: &K8sNodeUid, +) -> Result { + if caller_uid != &grant.uploader_node_uid { + return Err(RecoveryError::Refused( + "caller is not the approved recovery uploader", + )); + } + let artifact = snapshot + .artifact + .as_ref() + .ok_or(RecoveryError::Refused("recovery attempt has no artifact"))?; + if snapshot.sandbox_id != grant.sandbox_id + || artifact.local_artifact_id() != Some(grant.local_artifact_id) + || artifact.kind() != ArtifactKind::Full + || artifact.origin() != ArtifactOrigin::Suspend + || artifact.snapshot_name().is_some() + || snapshot.execution_metadata.as_ref() != Some(&grant.manifest.execution) + { + return Err(RecoveryError::Refused( + "recovery attempt differs from the approved artifact or execution metadata", + )); + } + if let Some(active) = grant.active_attempt_id.filter(|id| *id != snapshot.id) { + let failed: Option = sqlx::query_scalar(sql!( + "SELECT state='failed' FROM snapshots WHERE id=$1 FOR SHARE" + )) + .bind(active) + .fetch_optional(&mut **tx) + .await?; + if failed != Some(true) { + return Err(RecoveryError::Refused( + "recovery already has an active attempt", + )); + } + } + snapshot.execution_metadata = Some(grant.manifest.execution.clone()); + snapshot.vcpu = Some(grant.manifest.vcpu); + snapshot.memory_mib = Some(grant.manifest.memory_mib); + snapshot.template_id = Some(grant.manifest.template_id.clone()); + snapshot.content_hash = grant + .manifest + .content_hash + .clone() + .map(SnapshotContentHash::from_stored); + snapshot.source_node_id = Some(grant.uploader_node_id); + if grant.consumed_attempt_id == Some(snapshot.id) { + let persisted = get(&mut **tx, snapshot.id) + .await? + .ok_or(RecoveryError::Refused( + "consumed recovery attempt is absent", + ))?; + if persisted.state().is_ready() + && persisted.recovery_grant_id() == Some(grant.id) + && same_snapshot_identity(&persisted, &snapshot) + && persisted.source_node_uid() == Some(caller_uid) + && persisted.storage_location().is_some_and(|location| { + location.storage_connection_string() == snapshot.storage_uri.as_wire() + }) + { + return Ok(persisted); + } + return Err(RecoveryError::Refused( + "consumed recovery identity conflicts with replay", + )); + } + grant.validate_live(tx, caller_uid).await?; + let (persisted, outcome) = insert(tx, &snapshot).await?; + if outcome == InsertOutcome::Created { + let changed=sqlx::query(sql!("UPDATE snapshots SET generation=$2,recovery_grant_id=$3,source_node_uid=$4,timestamp=$5,last_attempt_at=NOW() + WHERE id=$1 AND recovery_grant_id IS NULL AND source_node_uid IS NULL AND state='creating' AND source_node_id=$6")) + .bind(snapshot.id).bind(grant.execution_generation).bind(grant.id).bind(&grant.uploader_node_uid).bind(grant.captured_at).bind(grant.uploader_node_id).execute(&mut **tx).await?; + if changed.rows_affected() != 1 { + return Err(RecoveryError::Refused("recovery registration changed")); + } + } else if persisted.recovery_grant_id() != Some(grant.id) + || persisted.source_node_uid() != Some(caller_uid) + { + return Err(RecoveryError::Refused( + "attempt is bound to another authority", + )); + } + sqlx::query(sql!( + "UPDATE backup_recovery_grants SET active_attempt_id=$2 WHERE id=$1" + )) + .bind(grant.id) + .bind(snapshot.id) + .execute(&mut **tx) + .await?; + get(&mut **tx, snapshot.id) + .await? + .ok_or(RecoveryError::Refused("recovery attempt disappeared")) +} + +/// Must commit atomically with Ready. Accepted upload work may finish after a +/// revoke, but it can never publish using revoked authority. +pub async fn consume( + conn: &mut sqlx::PgConnection, + grant: &RecoveryGrant, + attempt: SnapshotId, +) -> Result<(), RecoveryError> { + let result=sqlx::query(sql!("UPDATE backup_recovery_grants SET consumed_attempt_id=$2 WHERE id=$1 AND active_attempt_id=$2 AND consumed_attempt_id IS NULL AND revoked_at IS NULL AND expires_at>clock_timestamp()")) + .bind(grant.id).bind(attempt).execute(conn).await?; + if result.rows_affected() != 1 { + return Err(RecoveryError::Refused( + "recovery approval changed before finalization", + )); + } + Ok(()) +} + +/// Read-only discovery after a lost admin response; audit records outlive attempts. +#[derive(Debug, FromRow, Serialize)] +pub struct RecoveryApprovalSummary { + pub id: RecoveryGrantId, + pub local_artifact_id: LocalArtifactId, + pub execution_generation: i64, + pub uploader_node_id: NodeId, + pub uploader_node_uid: K8sNodeUid, + pub approved_at: String, + pub captured_at: String, + pub capture_evidence: Json, + pub expires_at: String, + pub active_attempt_id: Option, + pub consumed_attempt_id: Option, + pub revoked: bool, +} +pub async fn list_approvals( + conn: impl sqlx::PgExecutor<'_>, + sandbox: SandboxId, +) -> Result, sqlx::Error> { + sqlx::query_as(sql!("SELECT id,local_artifact_id,execution_generation,uploader_node_id,uploader_node_uid,approved_at::TEXT,captured_at::TEXT,capture_evidence,expires_at::TEXT,active_attempt_id,consumed_attempt_id,revoked_at IS NOT NULL AS revoked FROM backup_recovery_grants WHERE sandbox_id=$1 ORDER BY approved_at DESC,id DESC LIMIT 100")) + .bind(sandbox).fetch_all(conn).await +} + +/// Verify exact bytes and object bindings before Ready. A consumed replay needs +/// identical proof but may occur after restore changes the sandbox's placement. +pub async fn validate_ready( + conn: &mut sqlx::PgConnection, + grant: &RecoveryGrant, + attempt: &SnapshotRecord, + proofs: &[SnapshotArtifactDigest], + evidence: &[upload_evidence::ArtifactUploadEvidence], + bytes: i64, +) -> Result { + let vm_proofs: Vec<_> = proofs + .iter() + .copied() + .filter(|proof| proof.artifact != SnapshotArtifactFile::Metadata) + .collect(); + let metadata = evidence + .iter() + .find(|entry| entry.artifact == SnapshotArtifactFile::Metadata) + .and_then(|entry| entry.metadata_binding_digest); + // Catalog bytes describe compressed/encrypted objects, while the manifest + // describes logical files. Compare each count in its own representation. + let wire_bytes = evidence + .iter() + .filter(|entry| entry.artifact != SnapshotArtifactFile::Metadata) + .try_fold(0_i64, |sum, entry| { + i64::try_from(entry.wire_size_bytes) + .ok() + .and_then(|size| sum.checked_add(size)) + }); + if wire_bytes != Some(bytes) + || proofs.len() != vm_proofs.len() + 1 + || proofs.iter().any(|proof| proof.size_bytes == 0) + || !grant.manifest.matches(&vm_proofs, metadata) + || evidence.len() != proofs.len() + { + return Err(RecoveryError::Refused( + "uploaded files differ from the approved manifest or lack exact evidence", + )); + } + let location = attempt.storage_location().ok_or(RecoveryError::Refused( + "recovery attempt has no storage location", + ))?; + let uri = ExactStorageUri::parse(location.storage_connection_string()) + .map_err(|_| RecoveryError::Refused("recovery storage location is invalid"))?; + for proof in proofs { + let Some(object) = evidence + .iter() + .find(|entry| entry.artifact == proof.artifact) + else { + return Err(RecoveryError::Refused( + "uploaded file has no object evidence", + )); + }; + let expected = format!( + "{}/{}", + uri.parsed().key_prefix.trim_end_matches('/'), + proof.artifact.file_name() + ); + if object.object_key != expected || object.wire_size_bytes == 0 { + return Err(RecoveryError::Refused( + "uploaded object does not belong to this attempt", + )); + } + } + if grant.consumed_attempt_id == Some(attempt.id()) && attempt.state().is_ready() { + return Ok(false); + } + if grant.active_attempt_id != Some(attempt.id()) { + return Err(RecoveryError::Refused( + "recovery attempt no longer owns its approval", + )); + } + grant + .validate_live( + conn, + attempt.source_node_uid().ok_or(RecoveryError::Refused( + "recovery attempt has no uploader identity", + ))?, + ) + .await?; + Ok(true) +} + +#[cfg(test)] +mod tests; diff --git a/crates/libs/snapshot/store/src/recovery/tests.rs b/crates/libs/snapshot/store/src/recovery/tests.rs new file mode 100644 index 0000000000..1fa67936ba --- /dev/null +++ b/crates/libs/snapshot/store/src/recovery/tests.rs @@ -0,0 +1,63 @@ +use super::*; +use pplx_space_common_core::types::{Arch, ExecutionMetadata}; + +#[test] +fn approval_requires_a_sealed_unambiguous_full_artifact() { + let manifest = RecoveryManifest { + validation: SealedValidation { + sealed_full_suspend: true, + evidence_reference: "synthetic verification".into(), + }, + execution: SnapshotExecutionMetadata::new(ExecutionMetadata::initial_firecracker( + Arch::Aarch64, + vec![], + )) + .unwrap(), + metadata_binding_digest: format!("sha256:{}", "11".repeat(32)).parse().unwrap(), + vcpu: 2, + memory_mib: 1024, + template_id: "synthetic-template".parse().unwrap(), + content_hash: None, + artifacts: [ + SnapshotArtifactFile::Rootfs, + SnapshotArtifactFile::Memory, + SnapshotArtifactFile::VmState, + ] + .into_iter() + .map(|artifact| SnapshotArtifactDigest { + artifact, + digest: format!("sha256:{}", "22".repeat(32)).parse().unwrap(), + size_bytes: 32, + }) + .collect(), + }; + manifest.validate().unwrap(); + for case in [ + "unsealed", + "no-validation", + "missing-file", + "duplicate-role", + "generated-btrfs", + "empty-file", + "overflow", + "no-resources", + ] { + let mut invalid = manifest.clone(); + match case { + "unsealed" => invalid.validation.sealed_full_suspend = false, + "no-validation" => invalid.validation.evidence_reference.clear(), + "missing-file" => { + invalid.artifacts.pop(); + } + "duplicate-role" => invalid.artifacts[0].artifact = SnapshotArtifactFile::Memory, + "generated-btrfs" => { + invalid.artifacts[0].artifact = SnapshotArtifactFile::RootfsBtrfsSend + } + "empty-file" => invalid.artifacts[0].size_bytes = 0, + "overflow" => invalid.artifacts[0].size_bytes = i64::MAX as u64, + "no-resources" => invalid.vcpu = 0, + _ => unreachable!(), + } + assert!(invalid.validate().is_err(), "{case}"); + } +} diff --git a/docs/backup-recovery.md b/docs/backup-recovery.md new file mode 100644 index 0000000000..4ecc73e08b --- /dev/null +++ b/docs/backup-recovery.md @@ -0,0 +1,76 @@ +# Retained full-suspend recovery approvals + +A recovery approval authorizes publication of an already sealed, validated full +suspend from a specific Kubernetes worker incarnation. It does not seal an +incomplete capture, repair devices, start an upload, clear quarantine, change +placement, or wake the sandbox. Preserve the original retained artifact while +using independently owned staging copies for uploads. + +Deploy the additive schema first. Deploy token-aware placement writers, backup +identity enforcement, and evidence-producing upload workers before enabling the +`space-backup-recovery-grants-enabled` flag. Issuance defaults off. Every reader +that admits a backup must understand grant provenance before the first approval: +rolling back to a reader that treats grant-backed attempts as ordinary retry +authority is unsafe. Keep the audit schema on application rollback. Never run +its destructive down migration after issuing approvals. + +The managing organization's administrator can use these admin API routes: + +- `POST /sandboxes/{sandbox_id}/backup-recovery-grants` approves a manifest. +- `GET /sandboxes/{sandbox_id}/backup-recovery-grants` lists the newest 100 + approvals, including their active/consumed attempt and revocation state. Use + this after a lost response before retrying approval creation. +- `POST /sandboxes/{sandbox_id}/backup-recovery-grants/{grant_id}/revoke` revokes + an unconsumed approval and records the actor and reason. + +The admin OpenAPI document describes the request. Approval requires an exact +artifact and generation, uploader node ID and Node UID, capture time within the +recorded execution lifetime, capture evidence, reason, and an expiry within +24 hours. The manifest includes an explicit sealed-validation reference, captured +execution metadata and resource shape, template identity, stable metadata binding, +and logical SHA-256/size for every VM artifact. It accepts a complete raw full +suspend or the encrypted snapshot-volume/rootfs-COW pair. Missing, duplicate, +empty, or generated btrfs-send entries are rejected before an approval can +permit upload admission. + +Generate the stable metadata digest with the shared +`snapshot_metadata_binding` contract: it hashes canonical version-2 metadata, +excluding generated timestamps, measured sizes, and the local rootfs-send source +ID. Logical file sizes describe uncompressed plaintext inputs; catalog upload +size describes compressed/encrypted VM objects and excludes the metadata object. +Workers report both independently, alongside attempt-specific object keys and +available ETag/version identity. ETags are identity tokens, not content hashes. + +| Event | Catalog and approval outcome | Progress owner | +| --- | --- | --- | +| Valid Begin | One active attempt bound to the approved uploader and capture time | Existing upload worker | +| Concurrent second Begin | Refused while the first attempt is not failed | Operator or existing worker | +| Failed attempt before expiry | Failure recorded; a fresh attempt may reuse the live approval | Existing bounded retry policy | +| Expiry, revoke, changed state, newer materialization, or proof mismatch | Ready refused; Failed remains recordable | Operator inspects retained source and audit | +| Ready races revoke | Sandbox lock serializes them; exactly one wins | Winning transaction | +| Valid Ready | Logical/object evidence, Ready, and grant consumption commit together | Normal restore/lifecycle paths | +| Exact consumed replay | Same attempt and evidence remain idempotent | Caller retries lost response | +| New attempt after consumption | Cannot inherit ordinary retry authority, even after attempt pruning | New explicit authority required | + +An approval reserves subsequent registrations of its artifact for the approved +uploader, even when a failed ordinary upload records an earlier incarnation. +After expiry, revocation, or consumption, that older registration cannot bypass +the requirement for a new approval. Existing attempts retain their own terminal +reporting authority. + +An abandoned higher generation without a materialized pod is not supersession. +A higher active/materialized execution, a replaced source generation, newer +captured state, or a changed lifecycle/placement/artifact witness prevents +publication. Capture timestamps remain immutable across retries; attempt +liveness uses `last_attempt_at`. + +The administrator attests the source validation and capture provenance. Worker +hashes bind expected bytes on the honest authenticated-worker path; they do not +independently attest stored plaintext against a compromised uploader. A separate +object-reading verification service would be needed for that threat model. + +Upload retries retain their existing bounded deadlines. Expired or revoked grants +never extend them. At exhaustion, the operator owns the next decision; there is +no automatic grant renewal or source deletion. A successful Ready transition is +catalog evidence, not proof of a successful live restore. Validate restore on the +appropriate Linux/KVM substrate before enabling issuance in a deployment. diff --git a/docs/book/SUMMARY.md b/docs/book/SUMMARY.md index ec20b41fad..e25d35b99b 100644 --- a/docs/book/SUMMARY.md +++ b/docs/book/SUMMARY.md @@ -15,6 +15,7 @@ - [Troubleshooting](troubleshooting.md) - [Warm pools](warm-pools.md) - [Base snapshots](base-snapshots.md) + - [Retained snapshot recovery](backup-recovery.md) - [Internal RPC TLS](internal-rpc-tls.md) - [TLS interception exemptions](tls-interception-exemptions.md) - [Operator egress profiles](egress-profiles.md)