From 91c47c290bed0fef7599009784a3d2fa5d8d81ac Mon Sep 17 00:00:00 2001 From: viyatb-oai Date: Fri, 1 May 2026 17:21:45 -0700 Subject: [PATCH] fix: prune local security events Co-authored-by: Codex noreply@openai.com --- codex-rs/state/src/runtime.rs | 8 + codex-rs/state/src/runtime/security_events.rs | 264 ++++--------- .../src/runtime/security_events_tests.rs | 374 ++++++++++++++++++ 3 files changed, 458 insertions(+), 188 deletions(-) create mode 100644 codex-rs/state/src/runtime/security_events_tests.rs diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 223372fc00..5148cefdc4 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -101,6 +101,8 @@ pub use threads::ThreadFilterOptions; // metadata, rather than the exact sum of all persisted SQLite column bytes. const LOG_PARTITION_SIZE_LIMIT_BYTES: i64 = 10 * 1024 * 1024; const LOG_PARTITION_ROW_LIMIT: i64 = 1_000; +const SECURITY_EVENT_RETENTION_DAYS: i64 = 10; +const SECURITY_EVENT_PARTITION_ROW_LIMIT: i64 = 1_000; #[derive(Clone, Copy)] struct RuntimeDbSpec { @@ -309,6 +311,12 @@ impl StateRuntime { thread_updated_at_millis: Arc::new(AtomicI64::new(thread_updated_at_millis)), thread_recency_at_millis: Arc::new(AtomicI64::new(thread_recency_at_millis)), }); + if let Err(err) = runtime.run_security_events_startup_maintenance().await { + warn!( + "failed to run startup maintenance for state db at {}: {err}", + state_path.display(), + ); + } if let Err(err) = runtime.run_logs_startup_maintenance().await { warn!( "failed to run startup maintenance for logs db at {}: {err}", diff --git a/codex-rs/state/src/runtime/security_events.rs b/codex-rs/state/src/runtime/security_events.rs index 55b6a7dbbd..4c41d401d7 100644 --- a/codex-rs/state/src/runtime/security_events.rs +++ b/codex-rs/state/src/runtime/security_events.rs @@ -1,6 +1,4 @@ use super::*; -#[cfg(test)] -use crate::SecurityEventKind; use crate::SecurityEventResource; impl StateRuntime { @@ -9,6 +7,7 @@ impl StateRuntime { &self, event: &SecurityEventCreateParams, ) -> anyhow::Result<()> { + let mut tx = self.pool.begin().await?; sqlx::query( r#" INSERT INTO security_events ( @@ -55,8 +54,80 @@ INSERT INTO security_events ( .bind(event.reviewer.as_deref()) .bind(event.review_decision.as_deref()) .bind(event.details_json.as_deref()) - .execute(self.pool.as_ref()) + .execute(&mut *tx) .await?; + self.prune_security_events_after_insert(event, &mut tx) + .await?; + tx.commit().await?; + Ok(()) + } + + /// Keep local security-event history bounded while preserving recent context per bucket. + async fn prune_security_events_after_insert( + &self, + event: &SecurityEventCreateParams, + tx: &mut SqliteConnection, + ) -> anyhow::Result<()> { + let mut builder = QueryBuilder::::new( + r#" +DELETE FROM security_events +WHERE id IN ( + SELECT id + FROM ( + SELECT + id, + ROW_NUMBER() OVER ( + ORDER BY created_at DESC, id DESC + ) AS row_number + FROM security_events + WHERE + "#, + ); + if let Some(thread_id) = event.thread_id.as_deref() { + builder.push("thread_id = ").push_bind(thread_id); + } else { + builder.push("thread_id IS NULL"); + } + builder + .push(" AND kind = ") + .push_bind(event.kind.as_str()) + .push( + r#" + ) + WHERE row_number > + "#, + ) + .push_bind(SECURITY_EVENT_PARTITION_ROW_LIMIT) + .push("\n)"); + builder.build().execute(&mut *tx).await?; + Ok(()) + } + + pub(crate) async fn delete_security_events_before( + &self, + cutoff_ts: i64, + ) -> anyhow::Result { + let result = sqlx::query("DELETE FROM security_events WHERE created_at < ?") + .bind(cutoff_ts) + .execute(self.pool.as_ref()) + .await?; + Ok(result.rows_affected()) + } + + pub(crate) async fn run_security_events_startup_maintenance(&self) -> anyhow::Result<()> { + let Some(cutoff) = + Utc::now().checked_sub_signed(chrono::Duration::days(SECURITY_EVENT_RETENTION_DAYS)) + else { + return Ok(()); + }; + self.delete_security_events_before(cutoff.timestamp()) + .await?; + sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)") + .execute(self.pool.as_ref()) + .await?; + sqlx::query("PRAGMA incremental_vacuum") + .execute(self.pool.as_ref()) + .await?; Ok(()) } @@ -123,188 +194,5 @@ FROM security_events } #[cfg(test)] -mod tests { - use super::StateRuntime; - use super::test_support::unique_temp_dir; - use super::*; - use pretty_assertions::assert_eq; - - #[tokio::test] - async fn security_events_round_trip_and_filter() { - let codex_home = unique_temp_dir(); - let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string()) - .await - .expect("initialize runtime"); - - runtime - .insert_security_event(&SecurityEventCreateParams { - created_at: 1_700_000_000, - kind: SecurityEventKind::SandboxViolation, - thread_id: Some("thread-1".to_string()), - turn_id: Some("turn-1".to_string()), - call_id: Some("call-1".to_string()), - tool_name: Some("shell".to_string()), - resource: Some(SecurityEventResource::FileSystem), - sandbox_type: Some("macos_seatbelt".to_string()), - reason: Some("permission_denied".to_string()), - path: Some("/private/var/db".to_string()), - host: None, - port: None, - protocol: None, - method: None, - decision: None, - source: None, - review_id: None, - reviewer: None, - review_decision: None, - details_json: None, - }) - .await - .expect("insert filesystem event"); - runtime - .insert_security_event(&SecurityEventCreateParams { - created_at: 1_700_000_001, - kind: SecurityEventKind::SandboxViolation, - thread_id: Some("thread-1".to_string()), - turn_id: Some("turn-2".to_string()), - call_id: Some("call-2".to_string()), - tool_name: Some("shell".to_string()), - resource: Some(SecurityEventResource::Network), - sandbox_type: None, - reason: Some("not_allowed".to_string()), - path: None, - host: Some("example.com".to_string()), - port: Some(443), - protocol: Some("https_connect".to_string()), - method: Some("CONNECT".to_string()), - decision: Some("deny".to_string()), - source: Some("proxy_state".to_string()), - review_id: None, - reviewer: None, - review_decision: None, - details_json: None, - }) - .await - .expect("insert network event"); - runtime - .insert_security_event(&SecurityEventCreateParams { - created_at: 1_700_000_002, - kind: SecurityEventKind::AutoReviewDecision, - thread_id: Some("thread-2".to_string()), - turn_id: Some("turn-3".to_string()), - call_id: Some("call-3".to_string()), - tool_name: Some("shell".to_string()), - resource: None, - sandbox_type: None, - reason: None, - path: None, - host: None, - port: None, - protocol: None, - method: None, - decision: None, - source: None, - review_id: Some("review-1".to_string()), - reviewer: Some("auto_review".to_string()), - review_decision: Some("denied".to_string()), - details_json: None, - }) - .await - .expect("insert auto review event"); - - assert_eq!( - runtime - .list_security_events(&SecurityEventQuery { - thread_id: Some("thread-1".to_string()), - kind: Some(SecurityEventKind::SandboxViolation), - limit: None, - }) - .await - .expect("list sandbox events"), - vec![ - SecurityEvent { - id: 2, - created_at: 1_700_000_001, - kind: SecurityEventKind::SandboxViolation, - thread_id: Some("thread-1".to_string()), - turn_id: Some("turn-2".to_string()), - call_id: Some("call-2".to_string()), - tool_name: Some("shell".to_string()), - resource: Some(SecurityEventResource::Network), - sandbox_type: None, - reason: Some("not_allowed".to_string()), - path: None, - host: Some("example.com".to_string()), - port: Some(443), - protocol: Some("https_connect".to_string()), - method: Some("CONNECT".to_string()), - decision: Some("deny".to_string()), - source: Some("proxy_state".to_string()), - review_id: None, - reviewer: None, - review_decision: None, - details_json: None, - }, - SecurityEvent { - id: 1, - created_at: 1_700_000_000, - kind: SecurityEventKind::SandboxViolation, - thread_id: Some("thread-1".to_string()), - turn_id: Some("turn-1".to_string()), - call_id: Some("call-1".to_string()), - tool_name: Some("shell".to_string()), - resource: Some(SecurityEventResource::FileSystem), - sandbox_type: Some("macos_seatbelt".to_string()), - reason: Some("permission_denied".to_string()), - path: Some("/private/var/db".to_string()), - host: None, - port: None, - protocol: None, - method: None, - decision: None, - source: None, - review_id: None, - reviewer: None, - review_decision: None, - details_json: None, - }, - ] - ); - - assert_eq!( - runtime - .list_security_events(&SecurityEventQuery { - thread_id: None, - kind: Some(SecurityEventKind::AutoReviewDecision), - limit: Some(1), - }) - .await - .expect("list auto review events"), - vec![SecurityEvent { - id: 3, - created_at: 1_700_000_002, - kind: SecurityEventKind::AutoReviewDecision, - thread_id: Some("thread-2".to_string()), - turn_id: Some("turn-3".to_string()), - call_id: Some("call-3".to_string()), - tool_name: Some("shell".to_string()), - resource: None, - sandbox_type: None, - reason: None, - path: None, - host: None, - port: None, - protocol: None, - method: None, - decision: None, - source: None, - review_id: Some("review-1".to_string()), - reviewer: Some("auto_review".to_string()), - review_decision: Some("denied".to_string()), - details_json: None, - }] - ); - - let _ = tokio::fs::remove_dir_all(codex_home).await; - } -} +#[path = "security_events_tests.rs"] +mod tests; diff --git a/codex-rs/state/src/runtime/security_events_tests.rs b/codex-rs/state/src/runtime/security_events_tests.rs new file mode 100644 index 0000000000..2158f1f2ca --- /dev/null +++ b/codex-rs/state/src/runtime/security_events_tests.rs @@ -0,0 +1,374 @@ +use super::StateRuntime; +use super::test_support::unique_temp_dir; +use super::*; +use crate::SecurityEventKind; +use pretty_assertions::assert_eq; + +#[tokio::test] +async fn security_events_round_trip_and_filter() { + let codex_home = unique_temp_dir(); + let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string()) + .await + .expect("initialize runtime"); + + runtime + .insert_security_event(&SecurityEventCreateParams { + created_at: 1_700_000_000, + kind: SecurityEventKind::SandboxViolation, + thread_id: Some("thread-1".to_string()), + turn_id: Some("turn-1".to_string()), + call_id: Some("call-1".to_string()), + tool_name: Some("shell".to_string()), + resource: Some(SecurityEventResource::FileSystem), + sandbox_type: Some("macos_seatbelt".to_string()), + reason: Some("permission_denied".to_string()), + path: Some("/private/var/db".to_string()), + host: None, + port: None, + protocol: None, + method: None, + decision: None, + source: None, + review_id: None, + reviewer: None, + review_decision: None, + details_json: None, + }) + .await + .expect("insert filesystem event"); + runtime + .insert_security_event(&SecurityEventCreateParams { + created_at: 1_700_000_001, + kind: SecurityEventKind::SandboxViolation, + thread_id: Some("thread-1".to_string()), + turn_id: Some("turn-2".to_string()), + call_id: Some("call-2".to_string()), + tool_name: Some("shell".to_string()), + resource: Some(SecurityEventResource::Network), + sandbox_type: None, + reason: Some("not_allowed".to_string()), + path: None, + host: Some("example.com".to_string()), + port: Some(443), + protocol: Some("https_connect".to_string()), + method: Some("CONNECT".to_string()), + decision: Some("deny".to_string()), + source: Some("proxy_state".to_string()), + review_id: None, + reviewer: None, + review_decision: None, + details_json: None, + }) + .await + .expect("insert network event"); + runtime + .insert_security_event(&SecurityEventCreateParams { + created_at: 1_700_000_002, + kind: SecurityEventKind::AutoReviewDecision, + thread_id: Some("thread-2".to_string()), + turn_id: Some("turn-3".to_string()), + call_id: Some("call-3".to_string()), + tool_name: Some("shell".to_string()), + resource: None, + sandbox_type: None, + reason: None, + path: None, + host: None, + port: None, + protocol: None, + method: None, + decision: None, + source: None, + review_id: Some("review-1".to_string()), + reviewer: Some("auto_review".to_string()), + review_decision: Some("denied".to_string()), + details_json: None, + }) + .await + .expect("insert auto review event"); + + assert_eq!( + runtime + .list_security_events(&SecurityEventQuery { + thread_id: Some("thread-1".to_string()), + kind: Some(SecurityEventKind::SandboxViolation), + limit: None, + }) + .await + .expect("list sandbox events"), + vec![ + SecurityEvent { + id: 2, + created_at: 1_700_000_001, + kind: SecurityEventKind::SandboxViolation, + thread_id: Some("thread-1".to_string()), + turn_id: Some("turn-2".to_string()), + call_id: Some("call-2".to_string()), + tool_name: Some("shell".to_string()), + resource: Some(SecurityEventResource::Network), + sandbox_type: None, + reason: Some("not_allowed".to_string()), + path: None, + host: Some("example.com".to_string()), + port: Some(443), + protocol: Some("https_connect".to_string()), + method: Some("CONNECT".to_string()), + decision: Some("deny".to_string()), + source: Some("proxy_state".to_string()), + review_id: None, + reviewer: None, + review_decision: None, + details_json: None, + }, + SecurityEvent { + id: 1, + created_at: 1_700_000_000, + kind: SecurityEventKind::SandboxViolation, + thread_id: Some("thread-1".to_string()), + turn_id: Some("turn-1".to_string()), + call_id: Some("call-1".to_string()), + tool_name: Some("shell".to_string()), + resource: Some(SecurityEventResource::FileSystem), + sandbox_type: Some("macos_seatbelt".to_string()), + reason: Some("permission_denied".to_string()), + path: Some("/private/var/db".to_string()), + host: None, + port: None, + protocol: None, + method: None, + decision: None, + source: None, + review_id: None, + reviewer: None, + review_decision: None, + details_json: None, + }, + ] + ); + + assert_eq!( + runtime + .list_security_events(&SecurityEventQuery { + thread_id: None, + kind: Some(SecurityEventKind::AutoReviewDecision), + limit: Some(1), + }) + .await + .expect("list auto review events"), + vec![SecurityEvent { + id: 3, + created_at: 1_700_000_002, + kind: SecurityEventKind::AutoReviewDecision, + thread_id: Some("thread-2".to_string()), + turn_id: Some("turn-3".to_string()), + call_id: Some("call-3".to_string()), + tool_name: Some("shell".to_string()), + resource: None, + sandbox_type: None, + reason: None, + path: None, + host: None, + port: None, + protocol: None, + method: None, + decision: None, + source: None, + review_id: Some("review-1".to_string()), + reviewer: Some("auto_review".to_string()), + review_decision: Some("denied".to_string()), + details_json: None, + }] + ); + + let _ = tokio::fs::remove_dir_all(codex_home).await; +} + +#[tokio::test] +async fn insert_security_event_prunes_old_rows_per_thread_and_kind() { + let codex_home = unique_temp_dir(); + let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string()) + .await + .expect("initialize runtime"); + + for offset in 0..=SECURITY_EVENT_PARTITION_ROW_LIMIT { + runtime + .insert_security_event(&SecurityEventCreateParams { + created_at: 1_700_000_000 + offset, + kind: SecurityEventKind::SandboxViolation, + thread_id: Some("thread-1".to_string()), + turn_id: None, + call_id: None, + tool_name: None, + resource: Some(SecurityEventResource::FileSystem), + sandbox_type: None, + reason: None, + path: None, + host: None, + port: None, + protocol: None, + method: None, + decision: None, + source: None, + review_id: None, + reviewer: None, + review_decision: None, + details_json: None, + }) + .await + .expect("insert sandbox event"); + } + runtime + .insert_security_event(&SecurityEventCreateParams { + created_at: 1_700_100_000, + kind: SecurityEventKind::AutoReviewDecision, + thread_id: Some("thread-1".to_string()), + turn_id: None, + call_id: None, + tool_name: None, + resource: None, + sandbox_type: None, + reason: None, + path: None, + host: None, + port: None, + protocol: None, + method: None, + decision: None, + source: None, + review_id: Some("review-1".to_string()), + reviewer: Some("auto_review".to_string()), + review_decision: Some("denied".to_string()), + details_json: None, + }) + .await + .expect("insert review event"); + runtime + .insert_security_event(&SecurityEventCreateParams { + created_at: 1_700_200_000, + kind: SecurityEventKind::SandboxViolation, + thread_id: Some("thread-2".to_string()), + turn_id: None, + call_id: None, + tool_name: None, + resource: Some(SecurityEventResource::Network), + sandbox_type: None, + reason: None, + path: None, + host: None, + port: None, + protocol: None, + method: None, + decision: None, + source: None, + review_id: None, + reviewer: None, + review_decision: None, + details_json: None, + }) + .await + .expect("insert other thread event"); + + let sandbox_events = runtime + .list_security_events(&SecurityEventQuery { + thread_id: Some("thread-1".to_string()), + kind: Some(SecurityEventKind::SandboxViolation), + limit: None, + }) + .await + .expect("list sandbox events"); + assert_eq!( + sandbox_events.len(), + usize::try_from(SECURITY_EVENT_PARTITION_ROW_LIMIT).expect("row limit fits usize") + ); + assert_eq!( + sandbox_events.last().map(|event| event.created_at), + Some(1_700_000_001) + ); + + assert_eq!( + runtime + .list_security_events(&SecurityEventQuery { + thread_id: Some("thread-1".to_string()), + kind: Some(SecurityEventKind::AutoReviewDecision), + limit: None, + }) + .await + .expect("list review events") + .len(), + 1 + ); + assert_eq!( + runtime + .list_security_events(&SecurityEventQuery { + thread_id: Some("thread-2".to_string()), + kind: Some(SecurityEventKind::SandboxViolation), + limit: None, + }) + .await + .expect("list other thread events") + .len(), + 1 + ); + + let _ = tokio::fs::remove_dir_all(codex_home).await; +} + +#[tokio::test] +async fn delete_security_events_before_removes_old_rows() { + let codex_home = unique_temp_dir(); + let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string()) + .await + .expect("initialize runtime"); + + for created_at in [1_700_000_000, 1_700_000_100] { + runtime + .insert_security_event(&SecurityEventCreateParams { + created_at, + kind: SecurityEventKind::SandboxViolation, + thread_id: Some("thread-1".to_string()), + turn_id: None, + call_id: None, + tool_name: None, + resource: Some(SecurityEventResource::FileSystem), + sandbox_type: None, + reason: None, + path: None, + host: None, + port: None, + protocol: None, + method: None, + decision: None, + source: None, + review_id: None, + reviewer: None, + review_decision: None, + details_json: None, + }) + .await + .expect("insert security event"); + } + + assert_eq!( + runtime + .delete_security_events_before(1_700_000_050) + .await + .expect("delete old events"), + 1 + ); + assert_eq!( + runtime + .list_security_events(&SecurityEventQuery { + thread_id: Some("thread-1".to_string()), + kind: Some(SecurityEventKind::SandboxViolation), + limit: None, + }) + .await + .expect("list remaining events") + .into_iter() + .map(|event| event.created_at) + .collect::>(), + vec![1_700_000_100] + ); + + let _ = tokio::fs::remove_dir_all(codex_home).await; +}