🛡️ Sentinel: [CRITICAL] Fix Path Traversal in session storage

.jules/sentinel.md

@@ -0,0 +1,4 @@
+## 2024-04-10 - Path Traversal in Session Storage
+**Vulnerability:** Path traversal existed in both the Python (`src/session_store.py`) and Rust (`rust/crates/runtime/src/session_control.rs`) implementations because unsanitized `session_id` strings were used directly in file paths.
+**Learning:** Both reference implementations lacked central validation logic for system identifiers derived from external/user input.
+**Prevention:** Always validate and restrict identifier parameters (like session IDs) by checking for explicit disallow-lists (like path separators `/`, `\`, and directory traversal markers `.`, `..`) before using them in file operations.

rust/crates/runtime/src/session_control.rs

@@ -74,13 +74,17 @@ impl SessionStore {
         &self.workspace_root
     }
 
-    #[must_use]
-    pub fn create_handle(&self, session_id: &str) -> SessionHandle {
+    pub fn create_handle(&self, session_id: &str) -> Result<SessionHandle, SessionControlError> {
+        if !is_valid_session_id(session_id) {
+            return Err(SessionControlError::Format(format!(
+                "Invalid session ID: {session_id}"
+            )));
+        }
         let id = session_id.to_string();
         let path = self
             .sessions_root
             .join(format!("{id}.{PRIMARY_SESSION_EXTENSION}"));
-        SessionHandle { id, path }
+        Ok(SessionHandle { id, path })
     }
 
     pub fn resolve_reference(&self, reference: &str) -> Result<SessionHandle, SessionControlError> {
@@ -116,6 +120,11 @@ impl SessionStore {
     }
 
     pub fn resolve_managed_path(&self, session_id: &str) -> Result<PathBuf, SessionControlError> {
+        if !is_valid_session_id(session_id) {
+            return Err(SessionControlError::Format(format!(
+                "Invalid session ID: {session_id}"
+            )));
+        }
         for extension in [PRIMARY_SESSION_EXTENSION, LEGACY_SESSION_EXTENSION] {
             let path = self.sessions_root.join(format!("{session_id}.{extension}"));
             if path.exists() {
@@ -223,7 +232,7 @@ impl SessionStore {
     ) -> Result<ForkedManagedSession, SessionControlError> {
         let parent_session_id = session.session_id.clone();
         let forked = session.fork(branch_name);
-        let handle = self.create_handle(&forked.session_id);
+        let handle = self.create_handle(&forked.session_id)?;
         let branch_name = forked
             .fork
             .as_ref()
@@ -343,12 +352,25 @@ pub fn create_managed_session_handle_for(
     base_dir: impl AsRef<Path>,
     session_id: &str,
 ) -> Result<SessionHandle, SessionControlError> {
+    if !is_valid_session_id(session_id) {
+        return Err(SessionControlError::Format(format!(
+            "Invalid session ID: {session_id}"
+        )));
+    }
     let id = session_id.to_string();
     let path =
         managed_sessions_dir_for(base_dir)?.join(format!("{id}.{PRIMARY_SESSION_EXTENSION}"));
     Ok(SessionHandle { id, path })
 }
 
+#[must_use]
+pub fn is_valid_session_id(session_id: &str) -> bool {
+    if session_id.is_empty() || session_id == "." || session_id.contains("..") {
+        return false;
+    }
+    !session_id.contains(['/', '\\'])
+}
+
 pub fn resolve_session_reference(reference: &str) -> Result<SessionHandle, SessionControlError> {
     resolve_session_reference_for(env::current_dir()?, reference)
 }
@@ -397,6 +419,11 @@ pub fn resolve_managed_session_path_for(
     base_dir: impl AsRef<Path>,
     session_id: &str,
 ) -> Result<PathBuf, SessionControlError> {
+    if !is_valid_session_id(session_id) {
+        return Err(SessionControlError::Format(format!(
+            "Invalid session ID: {session_id}"
+        )));
+    }
     let directory = managed_sessions_dir_for(base_dir)?;
     for extension in [PRIMARY_SESSION_EXTENSION, LEGACY_SESSION_EXTENSION] {
         let path = directory.join(format!("{session_id}.{extension}"));
@@ -713,7 +740,9 @@ mod tests {
         session
             .push_user_text(text)
             .expect("session message should save");
-        let handle = store.create_handle(&session.session_id);
+        let handle = store
+            .create_handle(&session.session_id)
+            .expect("handle should create");
         let session = session.with_persistence_path(handle.path.clone());
         session
             .save_to_path(&handle.path)

src/session_store.py

@@ -13,23 +13,36 @@ class StoredSession:
     output_tokens: int
 
 
-DEFAULT_SESSION_DIR = Path('.port_sessions')
+DEFAULT_SESSION_DIR = Path(".port_sessions")
+
+
+def validate_session_id(session_id: str) -> None:
+    if "/" in session_id or "\\" in session_id:
+        raise ValueError(f"Invalid session ID: contains path separators ({session_id})")
+    if session_id in (".", ".."):
+        raise ValueError(f"Invalid session ID: cannot be '.' or '..' ({session_id})")
+    if ".." in session_id:
+        raise ValueError(
+            f"Invalid session ID: contains directory traversal ('..') ({session_id})"
+        )
 
 
 def save_session(session: StoredSession, directory: Path | None = None) -> Path:
+    validate_session_id(session.session_id)
     target_dir = directory or DEFAULT_SESSION_DIR
     target_dir.mkdir(parents=True, exist_ok=True)
-    path = target_dir / f'{session.session_id}.json'
+    path = target_dir / f"{session.session_id}.json"
     path.write_text(json.dumps(asdict(session), indent=2))
     return path
 
 
 def load_session(session_id: str, directory: Path | None = None) -> StoredSession:
+    validate_session_id(session_id)
     target_dir = directory or DEFAULT_SESSION_DIR
-    data = json.loads((target_dir / f'{session_id}.json').read_text())
+    data = json.loads((target_dir / f"{session_id}.json").read_text())
     return StoredSession(
-        session_id=data['session_id'],
-        messages=tuple(data['messages']),
-        input_tokens=data['input_tokens'],
-        output_tokens=data['output_tokens'],
+        session_id=data["session_id"],
+        messages=tuple(data["messages"]),
+        input_tokens=data["input_tokens"],
+        output_tokens=data["output_tokens"],
     )