fix(3.6.8): auth resilience — circuit breaker pauses run on token revocation, preserves findings

- Add is_auth_failure() detector (401, OAuth revoked, session expired, invalid key)
- Circuit breaker: 3 consecutive auth failures auto-pause instead of burning 66 agents
- Auth-aware park_exhausted(): clear message + fallback provider switch via /continue
- No retry burn on auth errors (immediate return like exhaustion)
- Recon preserves HTTP probe facts when model auth fails
- REPL phase tracking: paused (auth) distinct from paused (quota)
- RELEASE.md updated with auth resilience section

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
CyberSecurityUP
2026-08-08 09:14:21 -03:00
co-authored by Claude Opus 4.6
parent e956b482b9
commit 3c49a83578
4 changed files with 134 additions and 23 deletions
+27 -1
View File
@@ -5,7 +5,33 @@
**License:** MIT **License:** MIT
**Credits:** Joas A Santos & Red Team Leaders **Credits:** Joas A Santos & Red Team Leaders
## v3.6.8 — Bugfix: Ollama error handling, empty-evidence validation, single-model warnings ## v3.6.8 — Auth resilience, circuit breaker, Ollama error handling, empty-evidence validation
### Auth Resilience & Circuit Breaker (NEW)
- **`is_auth_failure()` detector.** New function recognises OAuth token revocation
(401), session expiry, invalid/revoked API keys, and "not logged in" errors from
subscription CLIs. Distinct from `is_exhaustion()` (quota/rate-limit) — auth
failures are non-recoverable without re-login or provider switch.
- **Circuit breaker (3 consecutive auth failures → auto-pause).** A shared atomic
counter tracks consecutive auth failures across ALL agents. After 3 failures the
pool pauses the run BEFORE burning through the remaining agents on a dead token.
Previously, a revoked OAuth token caused all 66+ agents to silently return 0
findings with no pause or warning.
- **Auth-aware park: findings preserved, fallback offered.** When auth fails the
run parks with a clear message:
`⏸ authentication failed (...). Run is PAUSED — all findings so far are SAFE.`
The user can `/continue openai:gpt-5.1` (or any provider) to switch and resume.
All `LiveCheckpoint` findings on disk are preserved across the pause.
- **No retry burn on auth failure.** `one()` returns immediately on auth errors
instead of retrying 3 times against a dead token (same as quota exhaustion).
- **Recon preserves probe facts on auth failure.** When model recon fails with an
auth error, the HTTP probe data is still returned and the pipeline continues
with probe-only intelligence instead of silently dropping everything.
- **Phase tracking for auth pauses.** The REPL status line shows `paused (auth)`
(distinct from `paused (quota)`) so the operator knows the root cause at a glance.
### Bugfixes
- **Better Ollama/local provider error messages.** Connection-refused and timeout - **Better Ollama/local provider error messages.** Connection-refused and timeout
errors now name the provider, URL, and suggest checking if the server is running. errors now name the provider, URL, and suggest checking if the server is running.
+1
View File
@@ -68,6 +68,7 @@ impl RunLive {
if self.feed.len() > 200 { self.feed.remove(0); } if self.feed.len() > 200 { self.feed.remove(0); }
} }
if low.contains("token/quota exhausted") || low.contains("run is paused") { self.phase = "paused (quota)".into(); } if low.contains("token/quota exhausted") || low.contains("run is paused") { self.phase = "paused (quota)".into(); }
else if low.contains("authentication failed") || low.contains("auth failed") || low.contains("circuit breaker") { self.phase = "paused (auth)".into(); }
else if low.contains("resumed — retrying") { self.phase = "exploiting".into(); } else if low.contains("resumed — retrying") { self.phase = "exploiting".into(); }
else if low.starts_with("recon") || low.starts_with("ai-recon") || low.contains("recon round") || low.contains("intensity") || low.starts_with("probe:") { self.phase = "recon".into(); } else if low.starts_with("recon") || low.starts_with("ai-recon") || low.contains("recon round") || low.contains("intensity") || low.starts_with("probe:") { self.phase = "recon".into(); }
else if low.contains("selected") && low.contains("agent") { else if low.contains("selected") && low.contains("agent") {
+21 -3
View File
@@ -610,7 +610,12 @@ pub async fn run(cfg: RunConfig, lib: &Library, pool: &ModelPool, tx: Sender<Str
(ag.name.clone(), text, f) (ag.name.clone(), text, f)
} }
Err(e) => { Err(e) => {
let _ = txc.send(format!("exploit {} failed: {e}", ag.name)).await; let is_auth = crate::pool::is_auth_failure(&e);
if is_auth {
let _ = txc.send(format!("⚠ exploit {} auth failed — findings so far are SAFE, run is pausing: {e}", ag.name)).await;
} else {
let _ = txc.send(format!("exploit {} failed: {e}", ag.name)).await;
}
(ag.name.clone(), format!("ERROR: {e}"), vec![]) (ag.name.clone(), format!("ERROR: {e}"), vec![])
} }
} }
@@ -1900,7 +1905,16 @@ async fn deep_recon(cfg: &RunConfig, pool: &ModelPool, probe_facts: &str, tx: &S
let _ = tx.send(format!("recon: intensity {} — actively enumerating (installing tools as needed)…", intensity)).await; let _ = tx.send(format!("recon: intensity {} — actively enumerating (installing tools as needed)…", intensity)).await;
match pool.complete_routed(Task::Recon, "recon", RECON_SYS, &user).await { match pool.complete_routed(Task::Recon, "recon", RECON_SYS, &user).await {
Ok((m, t)) => { let _ = tx.send(format!("recon round 1 complete via {}", m.label())).await; accum.push_str(&format!("\n\nMODEL RECON (round 1):\n{t}")); } Ok((m, t)) => { let _ = tx.send(format!("recon round 1 complete via {}", m.label())).await; accum.push_str(&format!("\n\nMODEL RECON (round 1):\n{t}")); }
Err(e) => { let _ = tx.send(format!("recon round 1 failed ({e}) — probe facts only")).await; return accum; } Err(e) => {
let is_auth = crate::pool::is_auth_failure(&e);
if is_auth {
let _ = tx.send(format!("recon round 1 auth failed ({e}) — continuing with probe facts; run will pause before exploit phase")).await;
accum.push_str(&format!("\n\nMODEL RECON (round 1):\n{e}"));
} else {
let _ = tx.send(format!("recon round 1 failed ({e}) — probe facts only")).await;
}
return accum;
}
} }
// Follow-up expansion rounds — each digs further using what's known so far. // Follow-up expansion rounds — each digs further using what's known so far.
@@ -1921,7 +1935,11 @@ async fn deep_recon(cfg: &RunConfig, pool: &ModelPool, probe_facts: &str, tx: &S
if novel.len() > 20 { let _ = tx.send(format!("recon round {round} via {} — expanded surface", m.label())).await; accum.push_str(&format!("\n\nMODEL RECON (round {round}):\n{novel}")); } if novel.len() > 20 { let _ = tx.send(format!("recon round {round} via {} — expanded surface", m.label())).await; accum.push_str(&format!("\n\nMODEL RECON (round {round}):\n{novel}")); }
else { let _ = tx.send(format!("recon round {round}: no new surface — recon converged")).await; break; } else { let _ = tx.send(format!("recon round {round}: no new surface — recon converged")).await; break; }
} }
Err(e) => { let _ = tx.send(format!("recon round {round} failed ({e})")).await; break; } Err(e) => {
let is_auth = crate::pool::is_auth_failure(&e);
let _ = tx.send(format!("recon round {round} {} ({e})", if is_auth { "auth failed" } else { "failed" })).await;
break;
}
} }
} }
accum accum
+85 -19
View File
@@ -20,6 +20,24 @@ pub fn is_exhaustion(e: &anyhow::Error) -> bool {
.any(|k| s.contains(k)) .any(|k| s.contains(k))
} }
/// Does this error look like an **authentication / authorization failure**
/// (revoked OAuth, expired session, invalid API key) — distinct from transient
/// quota/rate issues? Auth failures are non-recoverable without re-login, so
/// the run should pause immediately and offer fallback providers.
pub fn is_auth_failure(e: &anyhow::Error) -> bool {
let s = format!("{e:#}").to_lowercase();
[
"401", "403", "unauthorized", "token has been revoked",
"token revoked", "access token", "oauth", "session expired",
"not authenticated", "not logged in", "please log in",
"please login", "invalid api key", "invalid_api_key",
"api key expired", "authentication failed", "failed to authenticate",
"run /login",
]
.iter()
.any(|k| s.contains(k))
}
/// Task type used by the model router to pick the best model for the step. /// Task type used by the model router to pick the best model for the step.
#[derive(Clone, Copy, Debug)] #[derive(Clone, Copy, Debug)]
pub enum Task { pub enum Task {
@@ -64,6 +82,10 @@ pub struct ModelPool {
/// Fallback models the user added via `/continue <provider:model>` while /// Fallback models the user added via `/continue <provider:model>` while
/// paused — tried first on the next attempt. /// paused — tried first on the next attempt.
fallback: Arc<Mutex<Vec<ModelRef>>>, fallback: Arc<Mutex<Vec<ModelRef>>>,
/// Circuit breaker: consecutive auth/exhaustion failures across agents.
/// When this exceeds `AUTH_FAIL_THRESHOLD`, the pool auto-pauses instead of
/// burning through the remaining agents on a dead token.
consecutive_auth_fails: Arc<std::sync::atomic::AtomicUsize>,
} }
impl ModelPool { impl ModelPool {
@@ -96,9 +118,20 @@ impl ModelPool {
paused: Arc::new(AtomicBool::new(false)), paused: Arc::new(AtomicBool::new(false)),
resume: Arc::new(Notify::new()), resume: Arc::new(Notify::new()),
fallback: Arc::new(Mutex::new(Vec::new())), fallback: Arc::new(Mutex::new(Vec::new())),
consecutive_auth_fails: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
} }
} }
/// Reset the consecutive auth-failure counter (called on any successful completion).
fn reset_auth_fails(&self) {
self.consecutive_auth_fails.store(0, std::sync::atomic::Ordering::Relaxed);
}
/// Increment the consecutive auth-failure counter and return the new count.
fn inc_auth_fails(&self) -> usize {
self.consecutive_auth_fails.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1
}
/// Attach a progress channel so the subscription CLI streams structured /// Attach a progress channel so the subscription CLI streams structured
/// activity (commands run, files read, tools called) live. /// activity (commands run, files read, tools called) live.
pub fn set_progress(&self, tx: tokio::sync::mpsc::Sender<String>) { pub fn set_progress(&self, tx: tokio::sync::mpsc::Sender<String>) {
@@ -146,20 +179,32 @@ impl ModelPool {
self.paused.load(Ordering::Relaxed) self.paused.load(Ordering::Relaxed)
} }
/// Park the run on token/quota exhaustion: keep ALL state, emit a notice, /// Consecutive auth/quota failures needed to trip the circuit breaker and
/// and wait until the user runs `/continue` (or cancels). Returns when the /// auto-pause the run. Low threshold: 3 consecutive failures on the same
/// run should retry (pause cleared) or give up (cancelled). /// provider is enough signal that the token is dead.
async fn park_exhausted(&self, err: &anyhow::Error) { const AUTH_FAIL_THRESHOLD: usize = 3;
/// Park the run on token/quota exhaustion or auth failure: keep ALL state,
/// emit a notice, and wait until the user runs `/continue` (or cancels).
/// Returns when the run should retry (pause cleared) or give up (cancelled).
async fn park_exhausted(&self, err: &anyhow::Error, is_auth: bool) {
self.paused.store(true, Ordering::Relaxed); self.paused.store(true, Ordering::Relaxed);
if let Some(tx) = self.progress() { if let Some(tx) = self.progress() {
let msg = format!("{err:#}"); let msg = format!("{err:#}");
let short = msg.lines().next().unwrap_or(&msg); let short = msg.lines().next().unwrap_or(&msg);
let _ = tx let notice = if is_auth {
.send(format!( format!(
"notify: ⏸ authentication failed ({}). Run is PAUSED — all findings so far are SAFE. \
Fix: /continue <provider:model> to switch provider, or re-login and /continue.",
short.chars().take(120).collect::<String>()
)
} else {
format!(
"notify: ⏸ token/quota exhausted ({}). Run is PAUSED — type /continue when your quota renews, or switch with /model <provider:model> then /continue.", "notify: ⏸ token/quota exhausted ({}). Run is PAUSED — type /continue when your quota renews, or switch with /model <provider:model> then /continue.",
short.chars().take(120).collect::<String>() short.chars().take(120).collect::<String>()
)) )
.await; };
let _ = tx.send(notice).await;
} }
while self.paused.load(Ordering::Relaxed) && !self.is_cancelled() { while self.paused.load(Ordering::Relaxed) && !self.is_cancelled() {
let notified = self.resume.notified(); let notified = self.resume.notified();
@@ -169,8 +214,9 @@ impl ModelPool {
} }
} }
if !self.is_cancelled() { if !self.is_cancelled() {
self.reset_auth_fails(); // user resumed, reset counter
if let Some(tx) = self.progress() { if let Some(tx) = self.progress() {
let _ = tx.send("notify: ▶ resumed — retrying exhausted step.".to_string()).await; let _ = tx.send("notify: ▶ resumed — retrying with updated credentials/model.".to_string()).await;
} }
} }
} }
@@ -212,9 +258,9 @@ impl ModelPool {
}; };
match r { match r {
Ok(t) => return Ok(t), Ok(t) => return Ok(t),
// Don't burn retries on exhaustion — surface it so the caller // Don't burn retries on exhaustion or auth failure — surface
// can park and let the user /continue. // immediately so the caller can park and let the user /continue.
Err(e) if is_exhaustion(&e) => return Err(e), Err(e) if is_exhaustion(&e) || is_auth_failure(&e) => return Err(e),
Err(e) => last = e, Err(e) => last = e,
} }
} }
@@ -233,6 +279,20 @@ impl ModelPool {
if self.is_cancelled() { if self.is_cancelled() {
return Err(anyhow!("cancelled")); return Err(anyhow!("cancelled"));
} }
// Circuit breaker: if we've seen N consecutive auth failures across
// agents, pause immediately — don't burn another agent on a dead token.
let fail_count = self.consecutive_auth_fails.load(std::sync::atomic::Ordering::Relaxed);
if fail_count >= Self::AUTH_FAIL_THRESHOLD && !self.is_cancelled() {
self.park_exhausted(
&anyhow!("circuit breaker: {} consecutive auth failures — token/session likely dead", fail_count),
true,
).await;
if self.is_cancelled() {
return Err(anyhow!("cancelled"));
}
// After resume, retry with potentially new fallback models.
continue;
}
// User-supplied fallback models (via /continue) are tried first. // User-supplied fallback models (via /continue) are tried first.
let mut order = self.route(task); let mut order = self.route(task);
if let Ok(fb) = self.fallback.lock() { if let Ok(fb) = self.fallback.lock() {
@@ -244,25 +304,31 @@ impl ModelPool {
} }
let mut last = anyhow!("no candidate models"); let mut last = anyhow!("no candidate models");
let mut exhausted = false; let mut exhausted = false;
let mut auth_failed = false;
for m in &order { for m in &order {
if self.is_cancelled() { if self.is_cancelled() {
return Err(anyhow!("cancelled")); return Err(anyhow!("cancelled"));
} }
match self.one(label, m, system, user).await { match self.one(label, m, system, user).await {
Ok(text) => return Ok((m.clone(), text)), Ok(text) => {
self.reset_auth_fails(); // success resets circuit breaker
return Ok((m.clone(), text));
}
Err(e) => { Err(e) => {
if is_exhaustion(&e) { if is_auth_failure(&e) {
auth_failed = true;
self.inc_auth_fails();
} else if is_exhaustion(&e) {
exhausted = true; exhausted = true;
} }
last = e; last = e;
} }
} }
} }
// Every candidate failed. If it was token/quota exhaustion, park the // Every candidate failed. Park the run (keeping all state) so the user
// run until the user runs /continue, then retry the whole order (now // can fix auth or wait for quota renewal, then /continue.
// including any fallback model they added). Otherwise, give up. if (auth_failed || exhausted) && !self.is_cancelled() {
if exhausted && !self.is_cancelled() { self.park_exhausted(&last, auth_failed).await;
self.park_exhausted(&last).await;
continue; continue;
} }
return Err(last); return Err(last);