treewide: fix security vulnerabilities and race conditions

- scheduler: use tokio::process with kill_on_drop for proper timeout handling
- config: validate job_id against path traversal (alphanumeric, underscore, hyphen only)
- config: reject retry.max=0 (must be at least 1)
- git: add --no-absolute-file-names to tar extraction
- git: fix TOCTOU race with atomic temp dir + rename
- git: check tar exit code and cleanup on failure
- git: replace unwrap() with proper error handling
- main: handle scheduler errors with 5s restart delay
- main: skip sync for running jobs to prevent workdir corruption
- main: persist job_handles across scheduler restarts

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: aster-void

src/config.rs

@@ -6,6 +6,22 @@ use std::collections::HashMap;
 use std::str::FromStr;
 use std::time::Duration;
 
+fn validate_job_id(id: &str) -> Result<()> {
+    if id.is_empty() {
+        anyhow::bail!("Job ID cannot be empty");
+    }
+    if !id
+        .chars()
+        .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
+    {
+        anyhow::bail!(
+            "Invalid job ID '{}': must contain only alphanumeric characters, underscores, and hyphens",
+            id
+        );
+    }
+    Ok(())
+}
+
 #[derive(Debug, Clone, Default, PartialEq)]
 pub enum TimezoneConfig {
     #[default]
@@ -116,6 +132,8 @@ pub fn parse_config(content: &str) -> Result<(RunnerConfig, Vec<Job>)> {
         .jobs
         .into_iter()
         .map(|(id, job)| {
+            validate_job_id(&id)?;
+
             let schedule_str = format!("{} *", job.schedule.cron);
             let schedule = Schedule::from_str(&schedule_str)
                 .map_err(|e| anyhow!("Invalid cron '{}' in job '{}': {}", job.schedule.cron, id, e))?;
@@ -128,6 +146,12 @@ pub fn parse_config(content: &str) -> Result<(RunnerConfig, Vec<Job>)> {
             let retry = job
                 .retry
                 .map(|r| {
+                    if r.max == 0 {
+                        anyhow::bail!(
+                            "Invalid retry.max '0' in job '{}': must be at least 1 (use no retry config to disable retries)",
+                            id
+                        );
+                    }
                     let delay = parse_duration(&r.delay)
                         .map_err(|e| anyhow!("Invalid retry delay '{}' in job '{}': {}", r.delay, id, e))?;
                     let jitter = r.jitter
@@ -469,4 +493,55 @@ jobs:
         assert_eq!(parse_duration("500ms").unwrap(), Duration::from_millis(500));
         assert_eq!(parse_duration("1000ms").unwrap(), Duration::from_millis(1000));
     }
+
+    #[test]
+    fn reject_invalid_job_id() {
+        let yaml = r#"
+jobs:
+  "../escape":
+    schedule:
+      cron: "* * * * *"
+    run: echo test
+"#;
+        assert!(parse_config(yaml).is_err());
+    }
+
+    #[test]
+    fn reject_job_id_with_slash() {
+        let yaml = r#"
+jobs:
+  "foo/bar":
+    schedule:
+      cron: "* * * * *"
+    run: echo test
+"#;
+        assert!(parse_config(yaml).is_err());
+    }
+
+    #[test]
+    fn accept_valid_job_id() {
+        let yaml = r#"
+jobs:
+  my-job_123:
+    schedule:
+      cron: "* * * * *"
+    run: echo test
+"#;
+        let (_, jobs) = parse_config(yaml).unwrap();
+        assert_eq!(jobs[0].id, "my-job_123");
+    }
+
+    #[test]
+    fn reject_retry_max_zero() {
+        let yaml = r#"
+jobs:
+  test:
+    schedule:
+      cron: "* * * * *"
+    run: echo test
+    retry:
+      max: 0
+"#;
+        assert!(parse_config(yaml).is_err());
+    }
 }

src/git.rs

@@ -1,11 +1,13 @@
-use anyhow::Result;
+use anyhow::{Context, Result};
 use std::path::{Path, PathBuf};
 use std::process::Command;
 
 /// Ensures repo is cloned/synced to cache. Returns cache path.
 pub fn ensure_repo(source: &str) -> Result<PathBuf> {
     let cache_dir = get_cache_dir(source)?;
-    std::fs::create_dir_all(cache_dir.parent().unwrap())?;
+    if let Some(parent) = cache_dir.parent() {
+        std::fs::create_dir_all(parent)?;
+    }
 
     if cache_dir.exists() {
         sync_repo(source, &cache_dir)?;
@@ -25,8 +27,11 @@ fn is_remote(source: &str) -> bool {
 
 fn clone_repo(source: &str, dest: &Path) -> Result<()> {
     if is_remote(source) {
+        let dest_str = dest
+            .to_str()
+            .context("Destination path contains invalid UTF-8")?;
         let output = Command::new("git")
-            .args(["clone", source, dest.to_str().unwrap()])
+            .args(["clone", source, dest_str])
             .output()?;
 
         if !output.status.success() {
@@ -73,13 +78,17 @@ fn sync_repo(source: &str, dest: &Path) -> Result<()> {
 fn rsync_local(source: &str, dest: &Path) -> Result<()> {
     std::fs::create_dir_all(dest)?;
 
+    let dest_str = dest
+        .to_str()
+        .context("Destination path contains invalid UTF-8")?;
     let output = Command::new("rsync")
         .args([
             "-a",
             "--delete",
-            "--exclude", ".git",
+            "--exclude",
+            ".git",
             &format!("{}/", source),
-            dest.to_str().unwrap(),
+            dest_str,
         ])
         .output()?;
 
@@ -121,16 +130,33 @@ pub fn get_job_dir(sot_path: &Path, job_id: &str) -> PathBuf {
         .unwrap_or_else(|| PathBuf::from("/tmp"))
         .join("rollcron");
 
-    let sot_name = sot_path.file_name().unwrap().to_str().unwrap();
+    let sot_name = sot_path
+        .file_name()
+        .and_then(|s| s.to_str())
+        .unwrap_or("unknown");
 
     cache_base.join(format!("{}@{}", sot_name, job_id))
 }
 
 pub fn sync_to_job_dir(sot_path: &Path, job_dir: &Path) -> Result<()> {
-    if job_dir.exists() {
-        std::fs::remove_dir_all(job_dir)?;
+    let sot_str = sot_path
+        .to_str()
+        .context("Source path contains invalid UTF-8")?;
+    let job_dir_str = job_dir
+        .to_str()
+        .context("Job directory path contains invalid UTF-8")?;
+
+    // Use atomic temp directory to avoid TOCTOU race condition
+    let temp_dir = job_dir.with_extension("tmp");
+    let temp_dir_str = temp_dir
+        .to_str()
+        .context("Temp directory path contains invalid UTF-8")?;
+
+    // Clean up any leftover temp directory
+    if temp_dir.exists() {
+        std::fs::remove_dir_all(&temp_dir)?;
     }
-    std::fs::create_dir_all(job_dir)?;
+    std::fs::create_dir_all(&temp_dir)?;
 
     // Check if .git exists (remote repos have it, local rsync'd repos don't)
     if sot_path.join(".git").exists() {
@@ -141,35 +167,61 @@ pub fn sync_to_job_dir(sot_path: &Path, job_dir: &Path) -> Result<()> {
             .output()?;
 
         if !archive.status.success() {
+            std::fs::remove_dir_all(&temp_dir)?;
             let stderr = String::from_utf8_lossy(&archive.stderr);
             anyhow::bail!("git archive failed: {}", stderr);
         }
 
-        let extract = Command::new("tar")
-            .args(["-x"])
-            .current_dir(job_dir)
+        // Extract with security flags to prevent path traversal
+        let mut extract = Command::new("tar")
+            .args(["--no-absolute-file-names", "-x"])
+            .current_dir(&temp_dir)
             .stdin(std::process::Stdio::piped())
             .spawn()?;
 
-        use std::io::Write;
-        extract.stdin.unwrap().write_all(&archive.stdout)?;
+        {
+            use std::io::Write;
+            let stdin = extract
+                .stdin
+                .as_mut()
+                .context("Failed to open tar stdin")?;
+            stdin.write_all(&archive.stdout)?;
+        }
+
+        let status = extract.wait()?;
+        if !status.success() {
+            std::fs::remove_dir_all(&temp_dir)?;
+            anyhow::bail!("tar extraction failed with exit code: {:?}", status.code());
+        }
     } else {
         // For non-git dirs (rsync'd local repos), use rsync
         let output = Command::new("rsync")
             .args([
                 "-a",
                 "--delete",
-                &format!("{}/", sot_path.to_str().unwrap()),
-                job_dir.to_str().unwrap(),
+                &format!("{}/", sot_str),
+                temp_dir_str,
             ])
             .output()?;
 
         if !output.status.success() {
+            std::fs::remove_dir_all(&temp_dir)?;
             let stderr = String::from_utf8_lossy(&output.stderr);
             anyhow::bail!("rsync failed: {}", stderr);
         }
     }
 
+    // Atomic swap: remove old, rename temp to target
+    if job_dir.exists() {
+        std::fs::remove_dir_all(job_dir)?;
+    }
+    std::fs::rename(&temp_dir, job_dir).with_context(|| {
+        format!(
+            "Failed to rename {} to {}",
+            temp_dir_str, job_dir_str
+        )
+    })?;
+
     Ok(())
 }
 

src/main.rs

@@ -4,10 +4,19 @@ mod scheduler;
 
 use anyhow::Result;
 use clap::Parser;
+use std::collections::HashMap;
 use std::path::PathBuf;
+use std::sync::Arc;
 use std::time::Duration;
+use tokio::sync::RwLock;
 use tokio::time::interval;
 
+/// Tracks running job counts (supports multiple concurrent instances)
+pub type RunningJobs = Arc<RwLock<HashMap<String, usize>>>;
+
+/// Tracks running job handles (persists across scheduler restarts)
+pub type JobHandles = Arc<tokio::sync::Mutex<HashMap<String, Vec<tokio::task::JoinHandle<()>>>>>;
+
 const CONFIG_FILE: &str = "rollcron.yaml";
 
 #[derive(Parser)]
@@ -44,13 +53,20 @@ async fn main() -> Result<()> {
     println!("[rollcron] Cache: {}", sot_path.display());
 
     let (initial_runner, initial_jobs) = load_config(&sot_path)?;
-    sync_job_dirs(&sot_path, &initial_jobs)?;
+    sync_job_dirs(&sot_path, &initial_jobs, &HashMap::new())?;
 
     let (tx, mut rx) = tokio::sync::watch::channel((initial_runner, initial_jobs));
 
+    // Shared map of currently running job IDs -> instance count
+    let running_jobs: RunningJobs = Arc::new(RwLock::new(HashMap::new()));
+
+    // Shared map of job handles (persists across scheduler restarts)
+    let job_handles: JobHandles = Arc::new(tokio::sync::Mutex::new(HashMap::new()));
+
     // Spawn auto-sync task
     let source_clone = source.clone();
     let pull_interval = args.pull_interval;
+    let running_jobs_clone = Arc::clone(&running_jobs);
     tokio::spawn(async move {
         let mut ticker = interval(Duration::from_secs(pull_interval));
         loop {
@@ -67,10 +83,12 @@ async fn main() -> Result<()> {
 
             match load_config(&sot) {
                 Ok((runner, jobs)) => {
-                    if let Err(e) = sync_job_dirs(&sot, &jobs) {
+                    let running = running_jobs_clone.read().await;
+                    if let Err(e) = sync_job_dirs(&sot, &jobs, &running) {
                         eprintln!("[rollcron] Failed to sync job dirs: {}", e);
                         continue;
                     }
+                    drop(running);
                     let _ = tx.send((runner, jobs));
                     println!("[rollcron] Synced job directories");
                 }
@@ -83,8 +101,22 @@ async fn main() -> Result<()> {
     loop {
         let (runner, jobs) = rx.borrow_and_update().clone();
         let sot = sot_path.clone();
+        let running_jobs_ref = Arc::clone(&running_jobs);
+        let job_handles_ref = Arc::clone(&job_handles);
         tokio::select! {
-            _ = scheduler::run_scheduler(jobs, sot, runner) => {}
+            result = scheduler::run_scheduler(jobs, sot, runner, running_jobs_ref, job_handles_ref) => {
+                match result {
+                    Ok(()) => {
+                        // Scheduler exited normally (shouldn't happen with infinite loop)
+                        eprintln!("[rollcron] Scheduler exited unexpectedly, restarting...");
+                    }
+                    Err(e) => {
+                        eprintln!("[rollcron] Scheduler error: {}", e);
+                        eprintln!("[rollcron] Restarting scheduler in 5 seconds...");
+                        tokio::time::sleep(Duration::from_secs(5)).await;
+                    }
+                }
+            }
             _ = rx.changed() => {
                 println!("[rollcron] Config updated, restarting scheduler");
             }
@@ -99,8 +131,19 @@ fn load_config(sot_path: &PathBuf) -> Result<(config::RunnerConfig, Vec<config::
     config::parse_config(&content)
 }
 
-fn sync_job_dirs(sot_path: &PathBuf, jobs: &[config::Job]) -> Result<()> {
+fn sync_job_dirs(
+    sot_path: &PathBuf,
+    jobs: &[config::Job],
+    running_jobs: &HashMap<String, usize>,
+) -> Result<()> {
     for job in jobs {
+        if running_jobs.get(&job.id).copied().unwrap_or(0) > 0 {
+            println!(
+                "[rollcron] Skipping sync for job '{}' (currently running)",
+                job.id
+            );
+            continue;
+        }
         let job_dir = git::get_job_dir(sot_path, &job.id);
         git::sync_to_job_dir(sot_path, &job_dir)?;
     }