fix(stream): close remaining ingress races

Author: schickling

src/agent_author.rs

@@ -229,7 +229,7 @@ fn author_stream(
             format!("acquire catalog-authoring lock: {error:#}"),
         )
     })?;
-    let found = crate::discover(catalog_root);
+    let found = crate::discover_strict(catalog_root);
     if let Some(error) = found.errors.first() {
         return Err(AuthorError::new(
             "catalog-malformed",
@@ -1949,4 +1949,34 @@ mod tests {
             "invalid-stream"
         );
     }
+
+    #[test]
+    fn stream_authoring_refuses_catalogs_with_concealed_declarations() {
+        use std::os::unix::fs::symlink;
+
+        let temporary = tempfile::tempdir().unwrap();
+        let root = temporary.path().join("catalog");
+        let concealed = temporary.path().join("concealed");
+        let declaration_path = write(
+            &root,
+            "h/worker/agent.kdl",
+            &declaration("worker", "h", None, "catalog"),
+        );
+        write(
+            &concealed,
+            "agent.kdl",
+            &declaration("shadow", "h", None, "catalog"),
+        );
+        symlink(&concealed, root.join("concealed-link")).unwrap();
+        let original = fs::read(&declaration_path).unwrap();
+
+        let error = add_stream(&root, "h.worker", "h", None, "events", None).unwrap_err();
+
+        assert_eq!(error.code(), "catalog-malformed");
+        assert!(
+            error.to_string().contains("unobservable declaration entry"),
+            "{error}"
+        );
+        assert_eq!(fs::read(declaration_path).unwrap(), original);
+    }
 }

src/event.rs

@@ -4,6 +4,8 @@
 //! own bounded, agent-local receipt ring rather than writing the agent's immutable Sent ledger.
 
 use std::fs::{self, File, OpenOptions};
+use std::io::Write as _;
+use std::os::fd::{AsRawFd as _, FromRawFd as _};
 use std::path::Path;
 use std::sync::atomic::{AtomicU64, Ordering};
 
@@ -263,7 +265,10 @@ pub fn emit(
                 record
                     .recent
                     .iter()
-                    .find(|entry| entry.key.as_deref() == key && entry.filename != filename)
+                    .find(|entry| {
+                        entry.filename != filename
+                            && key.is_none_or(|key| entry.key.as_deref() == Some(key))
+                    })
                     .map(|entry| entry.filename.clone())
             } else {
                 None
@@ -353,15 +358,63 @@ fn read_record(path: &Path) -> anyhow::Result<Option<StreamRecord>> {
 }
 
 fn write_record(path: &Path, record: &StreamRecord) -> anyhow::Result<()> {
+    use std::ffi::CString;
+
     let parent = path.parent().context("stream state has no parent")?;
     fs::create_dir_all(parent)?;
-    let temporary = parent.join(format!(
+    let directory = File::open(parent)
+        .with_context(|| format!("open stream state directory {}", parent.display()))?;
+    let temporary = format!(
         ".state.tmp-{}-{}",
         std::process::id(),
         TMP_COUNTER.fetch_add(1, Ordering::Relaxed)
-    ));
-    fs::write(&temporary, serde_json::to_vec(record)?)?;
-    fs::rename(&temporary, path)?;
+    );
+    let temporary = CString::new(temporary)?;
+    let target = CString::new(
+        path.file_name()
+            .context("stream state has no filename")?
+            .as_encoded_bytes(),
+    )?;
+    let fd = unsafe {
+        libc::openat(
+            directory.as_raw_fd(),
+            temporary.as_ptr(),
+            libc::O_WRONLY | libc::O_CREAT | libc::O_EXCL | libc::O_NOFOLLOW | libc::O_CLOEXEC,
+            0o600,
+        )
+    };
+    if fd < 0 {
+        return Err(std::io::Error::last_os_error()).with_context(|| {
+            format!(
+                "create fresh stream state temporary in {}",
+                parent.display()
+            )
+        });
+    }
+    let mut file = unsafe { File::from_raw_fd(fd) };
+    let result = (|| -> anyhow::Result<()> {
+        file.write_all(&serde_json::to_vec(record)?)?;
+        file.sync_all()?;
+        let renamed = unsafe {
+            libc::renameat(
+                directory.as_raw_fd(),
+                temporary.as_ptr(),
+                directory.as_raw_fd(),
+                target.as_ptr(),
+            )
+        };
+        if renamed != 0 {
+            return Err(std::io::Error::last_os_error()).context("publish stream state atomically");
+        }
+        directory.sync_all()?;
+        Ok(())
+    })();
+    if result.is_err() {
+        unsafe {
+            libc::unlinkat(directory.as_raw_fd(), temporary.as_ptr(), 0);
+        }
+    }
+    result?;
     Ok(())
 }
 

tests/event_e2e.rs

@@ -256,6 +256,45 @@ fn symlinked_stream_state_ancestor_cannot_escape_the_agent_capability() {
     assert_eq!(fs::read_dir(outside.path()).unwrap().count(), 0);
 }
 
+#[cfg(unix)]
+#[test]
+fn predictable_stream_state_temporary_symlink_is_never_followed() {
+    use std::os::unix::fs::symlink;
+
+    let catalog = tempfile::tempdir().unwrap();
+    let outside = tempfile::tempdir().unwrap();
+    let agent = declare_agent(catalog.path(), "\"running\"", "  stream \"gh-ci\" {}\n");
+    let state_dir = agent.join("resources/streams/gh-ci");
+    fs::create_dir_all(&state_dir).unwrap();
+    let victim = outside.path().join("victim");
+    fs::write(&victim, "must remain unchanged").unwrap();
+    for counter in 0..4096 {
+        symlink(
+            &victim,
+            state_dir.join(format!(".state.tmp-{}-{counter}", std::process::id())),
+        )
+        .unwrap();
+    }
+
+    let error = event::emit(
+        catalog.path(),
+        "hetz",
+        "hetz.worker",
+        "gh-ci",
+        "temp-symlink",
+        None,
+        None,
+        "payload",
+        false,
+    )
+    .unwrap_err()
+    .to_string();
+
+    assert!(error.contains("fresh stream state temporary"), "{error}");
+    assert_eq!(fs::read_to_string(victim).unwrap(), "must remain unchanged");
+    assert!(!state_dir.join("state.json").exists());
+}
+
 #[cfg(unix)]
 #[test]
 fn symlinked_inbox_cannot_escape_the_agent_capability() {
@@ -312,7 +351,7 @@ fn supersede_collapses_only_the_matching_key_and_preserves_archive_receipts() {
 fn keyless_supersede_replaces_the_stream_wide_head() {
     let catalog = tempfile::tempdir().unwrap();
     let agent = declare_agent(catalog.path(), "\"running\"", "  stream \"gh-ci\" {}\n");
-    let old = emit(catalog.path(), "old", None, true);
+    let old = emit(catalog.path(), "old", Some("pr-1"), true);
     let new = emit(catalog.path(), "new", None, true);
     assert_eq!(new.superseded.as_deref(), Some(old.filename.as_str()));
     assert!(!message::inbox_dir(&agent).join(old.filename).exists());