From f5b9d197a90159de3562c3f495fd6318fae5385d Mon Sep 17 00:00:00 2001 From: Mathieu JESER <6737167+bobdivx@users.noreply.github.com> Date: Thu, 22 Oct 2026 07:09:01 +0200 Subject: [PATCH] =?UTF-8?q?fix(cursor-worker):=20masquer=20les=20jetons=20?= =?UTF-8?q?des=20remotes=20et=20un=20process=20par=20d=C3=A9p=C3=B4t?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remotes Git : `https://user:token@host/…` → `https://***@host/…` dans le statut du worker (workdirs[].git_remote), ses logs, les sorties des scripts nœud et les logs git des déploiements. Helpers partagés redact_remote / redact_urls_in_text + tests. - Routage : le CLI exécute toute session dans le PREMIER --worker-dir, donc un agent lancé sur template-studio tournait dans Vigie. Le worker démarre maintenant un process par dépôt (même nom de machine, --data-dir propre), le premier garde la sonde /readyz. Arrêt / statut / scripts distants gèrent tous les process ; le statut expose repos_up / repo_workers. --- apps/server/src/cursor_worker.rs | 319 +++++++++++++++--- apps/server/src/cursor_worker_routes.rs | 9 +- apps/server/src/cursor_worker_tests.rs | 106 +++++- .../components/CursorWorkerSettingsPanel.tsx | 8 + apps/web/src/lib/api.ts | 3 + crates/deploy/src/lib.rs | 7 +- crates/shared/src/redact.rs | 79 +++++ 7 files changed, 475 insertions(+), 56 deletions(-) diff --git a/apps/server/src/cursor_worker.rs b/apps/server/src/cursor_worker.rs index 2e752e303..20e2dec6c 100644 --- a/apps/server/src/cursor_worker.rs +++ b/apps/server/src/cursor_worker.rs @@ -91,6 +91,35 @@ fn registered_path() -> PathBuf { state_dir().join("registered.json") } +/// Un sous-dossier par dépôt servi en plus du premier (pid, logs, data-dir). +fn repos_dir() -> PathBuf { + state_dir().join("repos") +} + +/// Identifiant stable d'un dépôt : `-`. +pub fn repo_slot_id(path: &str) -> String { + let base = Path::new(path.trim_end_matches('/')) + .file_name() + .and_then(|s| s.to_str()) + .unwrap_or("repo"); + let mut slug = slug_name(base); + slug.truncate(40); + if slug.is_empty() { + slug = "repo".into(); + } + // FNV-1a 32 bits : stable entre versions (contrairement à DefaultHasher). + let mut h: u32 = 0x811c_9dc5; + for b in path.as_bytes() { + h ^= u32::from(*b); + h = h.wrapping_mul(0x0100_0193); + } + format!("{slug}-{h:08x}") +} + +fn repo_slot_dir(path: &str) -> PathBuf { + repos_dir().join(repo_slot_id(path)) +} + /// HOME dédié au CLI installé par DevForge (persistant dans le volume data). pub fn managed_home() -> PathBuf { state_dir().join("home") @@ -413,7 +442,8 @@ fn read_git_remote(workdir: &Path) -> Option { return None; } let s = String::from_utf8_lossy(&out.stdout).trim().to_string(); - (!s.is_empty()).then_some(s) + // Un remote `https://x-access-token:…@github.com/…` ne sort jamais en clair. + (!s.is_empty()).then(|| devforge_shared::redact::redact_remote(&s)) } pub fn parse_extra_dirs(raw: &str) -> Vec { @@ -467,6 +497,29 @@ pub fn build_worker_args(name: &str, workdirs: &[String]) -> Result, Ok(args) } +/// Arguments d'un worker dédié à UN dépôt. +/// +/// Le CLI utilise le premier `--worker-dir` comme espace de travail de toute +/// session : un worker multi-dossiers exécute donc chaque agent dans le premier +/// dépôt, quel que soit le dépôt demandé. Un process par dépôt (même nom de +/// machine) laisse Cursor router chaque demande vers le bon checkout. Chaque +/// process a son propre `--data-dir` (verrou du CLI) et pas de sonde HTTP +/// (le port est réservé au worker principal). +pub fn build_repo_worker_args(name: &str, dir: &str, data_dir: &str) -> Vec { + let mut args = vec!["worker".to_string()]; + let name = name.trim(); + if !name.is_empty() { + args.push("--name".into()); + args.push(name.to_string()); + } + args.push("--worker-dir".into()); + args.push(dir.to_string()); + args.push("--data-dir".into()); + args.push(data_dir.to_string()); + args.push("start".into()); + args +} + /// Workdirs effectifs : repos éligibles, sinon un espace vide géré par DevForge /// (le worker démarre quand même et les repos s'ajoutent au prochain relancement). pub fn workdirs_or_fallback(paths: &[String]) -> Vec { @@ -534,6 +587,15 @@ pub fn stop_script(base: &ScriptBase) -> String { fi rm -f "$PIDF" fi +for f in "$D"/repos/*/worker.pid; do + [ -f "$f" ] || continue + P=$(cat "$f" 2>/dev/null) + if [ -n "$P" ] && kill -0 "$P" 2>/dev/null; then + kill -TERM "$P" 2>/dev/null; sleep 1 + kill -0 "$P" 2>/dev/null && kill -KILL "$P" 2>/dev/null + fi + rm -f "$f" +done echo "DF_RUNNING=0" "#, prelude = script_prelude(base) @@ -553,26 +615,46 @@ N=0 for d in {dirs}; do [ -n "$d" ] && [ -d "$d/.git" ] || continue [ "$N" -ge {max} ] && break - set -- "$@" --worker-dir "$d" + set -- "$@" "$d" N=$((N+1)) echo "DF_DIR=$d" done if [ "$N" -eq 0 ]; then mkdir -p "$D/workspace" - set -- --worker-dir "$D/workspace" + set -- "$D/workspace" fi command -v agent >/dev/null 2>&1 || {{ echo "CLI agent absent"; exit 5; }} SETSID="" command -v setsid >/dev/null 2>&1 && SETSID="setsid" -CURSOR_API_KEY={key} $SETSID nohup agent worker --name {name} "$@" --management-addr {mgmt} start "$D/worker.out" 2>"$D/worker.err" & +K={key} +FIRST="$1"; shift +# Un process par dépôt : le CLI exécute chaque session dans son premier --worker-dir. +CURSOR_API_KEY="$K" $SETSID nohup agent worker --name {name} --worker-dir "$FIRST" --management-addr {mgmt} start "$D/worker.out" 2>"$D/worker.err" & echo $! >"$PIDF" +I=1 +for d in "$@"; do + S="$D/repos/$I" + mkdir -p "$S/data" + CURSOR_API_KEY="$K" $SETSID nohup agent worker --name {name} --worker-dir "$d" --data-dir "$S/data" start "$S/worker.out" 2>"$S/worker.err" & + echo $! >"$S/worker.pid" + I=$((I+1)) +done sleep 4 P=$(cat "$PIDF") if kill -0 "$P" 2>/dev/null; then echo "DF_RUNNING=1" echo "DF_PID=$P" + UP=1 + for f in "$D"/repos/*/worker.pid; do + [ -f "$f" ] && kill -0 "$(cat "$f" 2>/dev/null)" 2>/dev/null && UP=$((UP+1)) + done + echo "DF_REPOS_UP=$UP" else rm -f "$PIDF" + for f in "$D"/repos/*/worker.pid; do + [ -f "$f" ] && kill -TERM "$(cat "$f" 2>/dev/null)" 2>/dev/null + rm -f "$f" + done echo "DF_RUNNING=0" echo "DF_LOGS_BEGIN" tail -n 30 "$D/worker.err" 2>/dev/null @@ -609,6 +691,11 @@ if [ -n "$P" ] && kill -0 "$P" 2>/dev/null; then else echo "DF_READY=0" fi + UP=1 + for f in "$D"/repos/*/worker.pid; do + [ -f "$f" ] && kill -0 "$(cat "$f" 2>/dev/null)" 2>/dev/null && UP=$((UP+1)) + done + echo "DF_REPOS_UP=$UP" else echo "DF_RUNNING=0" fi @@ -630,6 +717,8 @@ pub struct ScriptReport { pub pid: Option, pub ready: Option, pub dirs: Vec, + /// Process actifs (un par dépôt) après démarrage. + pub repos_up: Option, pub logs: String, } @@ -658,11 +747,12 @@ pub fn parse_script_output(out: &str) -> ScriptReport { "DF_PID" => r.pid = v.parse().ok(), "DF_READY" => r.ready = Some(v == "1"), "DF_DIR" => r.dirs.push(v.to_string()), + "DF_REPOS_UP" => r.repos_up = v.parse().ok(), "DF_STOPPED" | "DF_AGENT" | "DF_VERSION" => {} _ => logs.push(line), } } - r.logs = logs.join("\n").trim().to_string(); + r.logs = devforge_shared::redact::redact_urls_in_text(logs.join("\n").trim()); r } @@ -761,12 +851,62 @@ fn kill_pid(pid: u32) -> Result<(), String> { Ok(()) } +/// Pid d'un fichier pid, s'il pointe vers un process vivant. +fn live_pid_at(path: &Path) -> Option { + let raw = std::fs::read_to_string(path).ok()?; + let pid: u32 = raw.trim().parse().ok()?; + (pid > 0 && pid_is_alive(pid)).then_some(pid) +} + +/// Arrête les workers dédiés aux dépôts secondaires (dossiers `repos/*`). +/// Renvoie les pids arrêtés. +fn stop_repo_workers() -> Vec { + let mut stopped = Vec::new(); + let Ok(entries) = std::fs::read_dir(repos_dir()) else { + return stopped; + }; + for e in entries.flatten() { + let pidf = e.path().join("worker.pid"); + if let Some(pid) = live_pid_at(&pidf) { + if kill_pid(pid).is_ok() { + stopped.push(pid); + } + } + let _ = std::fs::remove_file(&pidf); + } + stopped +} + +/// État par dépôt servi : le premier est le worker principal, les autres ont +/// chacun leur process. +pub fn repo_workers_state(registered: &[String]) -> Vec { + registered + .iter() + .enumerate() + .map(|(i, path)| { + let pid = if i == 0 { + read_pid().filter(|p| pid_is_alive(*p)) + } else { + live_pid_at(&repo_slot_dir(path).join("worker.pid")) + }; + json!({ + "path": path, + "primary": i == 0, + "running": pid.is_some(), + "pid": pid, + }) + }) + .collect() +} + pub fn stop() -> Result { + let extra = stop_repo_workers(); let Some(pid) = read_pid() else { let _ = std::fs::remove_file(pid_path()); return Ok(json!({ "ok": true, "running": false, + "stopped_repo_workers": extra.len(), "message": "Aucun worker en cours", })); }; @@ -775,6 +915,7 @@ pub fn stop() -> Result { return Ok(json!({ "ok": true, "running": false, + "stopped_repo_workers": extra.len(), "message": "Process déjà arrêté", })); } @@ -784,26 +925,40 @@ pub fn stop() -> Result { "ok": true, "running": false, "stopped_pid": pid, + "stopped_repo_workers": extra.len(), "message": format!("Worker arrêté (pid {pid})"), })) } +fn tail_file(path: &Path, max_lines: usize) -> Option { + let raw = std::fs::read_to_string(path).ok()?; + let lines: Vec<&str> = raw.lines().collect(); + let start = lines.len().saturating_sub(max_lines); + (start < lines.len()).then(|| lines[start..].join("\n")) +} + pub fn read_logs_tail(max_lines: usize) -> String { let mut chunks = Vec::new(); for path in [out_path(), err_path()] { - if let Ok(raw) = std::fs::read_to_string(&path) { - let lines: Vec<&str> = raw.lines().collect(); - let start = lines.len().saturating_sub(max_lines); - if start < lines.len() { - chunks.push(format!( - "—— {} ——\n{}", - path.file_name().and_then(|s| s.to_str()).unwrap_or("log"), - lines[start..].join("\n") - )); - } + if let Some(tail) = tail_file(&path, max_lines) { + chunks.push(format!( + "—— {} ——\n{tail}", + path.file_name().and_then(|s| s.to_str()).unwrap_or("log"), + )); } } - chunks.join("\n\n") + // Dépôts secondaires arrêtés : dernière erreur, pour comprendre pourquoi. + let registered = load_registered(); + for path in registered.iter().skip(1) { + let dir = repo_slot_dir(path); + if live_pid_at(&dir.join("worker.pid")).is_some() { + continue; + } + if let Some(tail) = tail_file(&dir.join("worker.err"), 3) { + chunks.push(format!("—— {path} (arrêté) ——\n{tail}")); + } + } + devforge_shared::redact::redact_urls_in_text(&chunks.join("\n\n")) } pub fn load_registered() -> Vec { @@ -819,36 +974,23 @@ fn save_registered(paths: &[String]) -> Result<(), String> { std::fs::write(registered_path(), raw).map_err(|e| e.to_string()) } -/// Démarre avec un nom déjà résolu (nom du nœud par défaut). -pub fn start_named( - settings: &WorkerSettings, +/// Lance un process worker détaché (stdout/stderr vers fichiers) et le +/// récolte dans un thread pour éviter les zombies. +fn spawn_worker_process( bin: &Path, - name: &str, - workdirs: &[String], -) -> Result { - if is_running() { - return Err("Worker déjà en cours — arrête-le avant de relancer.".into()); - } - // Clé API si enregistrée, sinon session `cursor-agent login` (HOME géré). - let api_key = effective_api_key(settings).map(|(k, _)| k); - if api_key.is_none() && !is_logged_in(bin) { - return Err( - "Ni clé API Cursor ni connexion Cursor : utilise « Se connecter avec Cursor » ou colle une clé." - .into(), - ); - } - ensure_state_dir()?; - let workdirs = workdirs_or_fallback(workdirs); - let args = build_worker_args(name, &workdirs)?; - let out = std::fs::File::create(out_path()).map_err(|e| e.to_string())?; - let err = std::fs::File::create(err_path()).map_err(|e| e.to_string())?; - + args: &[String], + out: &Path, + err: &Path, + api_key: Option<&str>, +) -> Result { + let out = std::fs::File::create(out).map_err(|e| e.to_string())?; + let err = std::fs::File::create(err).map_err(|e| e.to_string())?; let mut cmd = Command::new(bin); - cmd.args(&args) + cmd.args(args) .stdin(Stdio::null()) .stdout(Stdio::from(out)) .stderr(Stdio::from(err)); - match &api_key { + match api_key { Some(k) => { cmd.env("CURSOR_API_KEY", k); } @@ -875,25 +1017,91 @@ pub fn start_named( .spawn() .map_err(|e| format!("Échec démarrage `{}`: {e}", bin.display()))?; let pid = child.id(); - std::fs::write(pid_path(), pid.to_string()).map_err(|e| e.to_string())?; - save_registered(&workdirs)?; - let _ = std::thread::Builder::new() .name("cursor-worker-reap".into()) .spawn(move || { let mut child = child; let _ = child.wait(); }); + Ok(pid) +} + +/// Démarre avec un nom déjà résolu (nom du nœud par défaut). +/// +/// Un process par dépôt : le premier dépôt garde le worker « principal » +/// (pid, logs, sonde `/readyz`), chaque dépôt suivant a le sien sous +/// `repos//`. Tous portent le même nom de machine : Cursor choisit le +/// process dont le dépôt correspond à la demande. +pub fn start_named( + settings: &WorkerSettings, + bin: &Path, + name: &str, + workdirs: &[String], +) -> Result { + if is_running() { + return Err("Worker déjà en cours — arrête-le avant de relancer.".into()); + } + // Restes d'un démarrage précédent (worker principal mort, secondaires vivants). + stop_repo_workers(); + // Clé API si enregistrée, sinon session `cursor-agent login` (HOME géré). + let api_key = effective_api_key(settings).map(|(k, _)| k); + if api_key.is_none() && !is_logged_in(bin) { + return Err( + "Ni clé API Cursor ni connexion Cursor : utilise « Se connecter avec Cursor » ou colle une clé." + .into(), + ); + } + ensure_state_dir()?; + let workdirs = workdirs_or_fallback(workdirs); + let args = build_worker_args(name, &workdirs[..1])?; + let pid = spawn_worker_process(bin, &args, &out_path(), &err_path(), api_key.as_deref())?; + std::fs::write(pid_path(), pid.to_string()).map_err(|e| e.to_string())?; + save_registered(&workdirs)?; + + let mut extra: Vec<(String, u32)> = Vec::new(); + let mut extra_errors: Vec = Vec::new(); + for dir in workdirs.iter().skip(1) { + let slot = repo_slot_dir(dir); + let data = slot.join("data"); + if let Err(e) = std::fs::create_dir_all(&data) { + extra_errors.push(format!("{dir} : {e}")); + continue; + } + let args = build_repo_worker_args(name, dir, &data.display().to_string()); + match spawn_worker_process( + bin, + &args, + &slot.join("worker.out"), + &slot.join("worker.err"), + api_key.as_deref(), + ) { + Ok(p) => { + let _ = std::fs::write(slot.join("worker.pid"), p.to_string()); + extra.push((dir.clone(), p)); + } + Err(e) => extra_errors.push(format!("{dir} : {e}")), + } + } - // Laisse le process s’initialiser un court instant. + // Laisse les process s’initialiser un court instant. std::thread::sleep(std::time::Duration::from_millis(600)); if !pid_is_alive(pid) { let logs = read_logs_tail(40); let _ = std::fs::remove_file(pid_path()); + stop_repo_workers(); return Err(format!( "Le worker s’est arrêté immédiatement. Vérifie la clé API et le réseau sortant (api2.cursor.sh).\n{logs}" )); } + for (dir, p) in &extra { + if !pid_is_alive(*p) { + let tail = tail_file(&repo_slot_dir(dir).join("worker.err"), 2).unwrap_or_default(); + extra_errors.push(devforge_shared::redact::redact_urls_in_text(&format!( + "{dir} : arrêté au démarrage {tail}" + ))); + } + } + let serving = workdirs.len() - extra_errors.len(); Ok(json!({ "ok": true, @@ -903,7 +1111,16 @@ pub fn start_named( "binary": bin.display().to_string(), "workdirs": workdirs, "workdirs_count": workdirs.len(), - "message": format!("Worker « {name} » démarré avec {} repo(s)", workdirs.len()), + "processes": 1 + extra.len(), + "repo_errors": extra_errors, + "message": if extra_errors.is_empty() { + format!("Worker « {name} » démarré : {serving} repo(s), un process par repo") + } else { + format!( + "Worker « {name} » démarré : {serving}/{} repo(s) actifs", + workdirs.len() + ) + }, })) } @@ -940,6 +1157,12 @@ pub async fn status_json(pool: &PgPool, node_name: &str) -> Result Result, headers: HeaderMap) -> Result {status.name}} /> + {status.running && status.repos_up != null ? ( + + ) : null} ; logs_tail: string; hint: string; }; diff --git a/crates/deploy/src/lib.rs b/crates/deploy/src/lib.rs index c2d511ef8..a4f9ab06d 100644 --- a/crates/deploy/src/lib.rs +++ b/crates/deploy/src/lib.rs @@ -868,7 +868,7 @@ impl DeployFacade { logs.push_str(&format!( "[git] exit={} {}\n", r.exit_code, - trim_out(&r.output) + devforge_shared::redact::redact_urls_in_text(&trim_out(&r.output)) )); if !r.ok { return DeployResult { @@ -879,7 +879,10 @@ impl DeployFacade { } } Err(e) => { - logs.push_str(&format!("[git] error: {e}\n")); + logs.push_str(&format!( + "[git] error: {}\n", + devforge_shared::redact::redact_urls_in_text(&e.to_string()) + )); return DeployResult { ok: false, git_sha: None, diff --git a/crates/shared/src/redact.rs b/crates/shared/src/redact.rs index 229013d0d..5bf4a5d79 100644 --- a/crates/shared/src/redact.rs +++ b/crates/shared/src/redact.rs @@ -286,6 +286,47 @@ pub fn redact_url(url: &str) -> String { format!("{scheme}{authority}{path}{query_and_frag}") } +/// Git remote for display: any userinfo (`user:token@`, `token@`) becomes +/// `***@`, so neither the token nor its account name leaks +/// (`https://x-access-token:ghp_…@github.com/o/r.git` → +/// `https://***@github.com/o/r.git`). SSH remotes (`git@host:o/r`) and URLs +/// without credentials are returned unchanged. +pub fn redact_remote(url: &str) -> String { + let t = url.trim(); + let Some(scheme_end) = t.find("://") else { + return t.to_string(); + }; + let (scheme, rest) = t.split_at(scheme_end + 3); + let authority_end = rest.find(['/', '?', '#']).unwrap_or(rest.len()); + match rest[..authority_end].rfind('@') { + Some(at) => format!("{scheme}{MASK}{}", &rest[at..]), + None => t.to_string(), + } +} + +/// Masks credentials of every URL found in free text (logs, CLI output): +/// `scheme://user:token@host` → `scheme://***@host`. The rest of the text is +/// unchanged. +pub fn redact_urls_in_text(s: &str) -> String { + let mut out = String::with_capacity(s.len()); + let mut rest = s; + while let Some(i) = rest.find("://") { + out.push_str(&rest[..i + 3]); + rest = &rest[i + 3..]; + let end = rest + .find(|c: char| { + c.is_whitespace() || matches!(c, '/' | '?' | '#' | '\'' | '"' | '<' | '>' | '`') + }) + .unwrap_or(rest.len()); + if let Some(at) = rest[..end].rfind('@') { + out.push_str(MASK); + rest = &rest[at..]; + } + } + out.push_str(rest); + out +} + fn is_sensitive_query_key(k: &str) -> bool { let n = normalise_key(k); is_sensitive_key(k) @@ -688,4 +729,42 @@ mod tests { assert_eq!(r["api_token"]["has_value"], true); assert_eq!(r["empty"]["has_value"], false); } + + #[test] + fn redact_remote_hides_userinfo() { + assert_eq!( + redact_remote("https://x-access-token:ghp_abc123@github.com/o/r.git"), + "https://***@github.com/o/r.git" + ); + assert_eq!( + redact_remote("https://ghp_abc@github.com/o/r"), + "https://***@github.com/o/r" + ); + assert_eq!( + redact_remote("https://github.com/o/r.git"), + "https://github.com/o/r.git" + ); + assert_eq!( + redact_remote("git@github.com:o/r.git"), + "git@github.com:o/r.git" + ); + assert_eq!( + redact_remote("http://u:p@host:8080/x@y"), + "http://***@host:8080/x@y" + ); + } + + #[test] + fn redact_urls_in_text_masks_every_url() { + let log = "fatal: unable to access 'https://x-access-token:ghp_secret@github.com/o/r.git/': 403\n\ + remote https://tok@gitlab.com/g/p ok, plain https://github.com/o/r"; + let r = redact_urls_in_text(log); + assert!(!r.contains("ghp_secret"), "{r}"); + assert!(!r.contains("tok@"), "{r}"); + assert!(r.contains("'https://***@github.com/o/r.git/'"), "{r}"); + assert!(r.contains("https://***@gitlab.com/g/p ok"), "{r}"); + assert!(r.contains("plain https://github.com/o/r"), "{r}"); + assert_eq!(redact_urls_in_text("no url here"), "no url here"); + assert_eq!(redact_urls_in_text("mail me@x.org"), "mail me@x.org"); + } }