feat(harness): add LLM dedup, budget warnings, agent nudge, image budget, and run resume

- LLM-based finding dedup: fast-path deterministic (title+endpoint, CWE+endpoint) + LLM semantic judge for fuzzy cases
- Progressive budget warnings at 70/85/95% with stop-new-agents gate at 95%
- Token tracking on ModelPool (record_tokens on every completion, set_token_limit from budget config)
- Agent lifecycle nudge: retry failed agents up to 3x with progressively simplified prompts
- Image budget enforcement: cap base64 screenshots in prompts, strip-all for recovery
- Run snapshot/resume: persist RunSnapshot to workdir, skip already-completed agents on resume
- 21 unit tests for all new features (500 total pass)
This commit is contained in:
CyberSecurityUP committed 2026-10-18 19:52:10 -03:00
1 parent 06715d11bd
commit 5af21f13f0
3 files changed
+520 -5

No files matched your search

@@ -952,6 +952,7 @@ async fn authz_matrix_pass(cfg: &RunConfig, recon: &str, tx: &Sender<String>) ->
pub async fn run(mut cfg: RunConfig, lib: &Library, pool: &ModelPool, tx: Sender<String>) -> RunOutput {
pool.set_progress(tx.clone());
pool.set_token_limit(cfg.budget.token_limit);
// A broad DEFAULT objective for a black-box web engagement when the operator
// set none — so a plain `/run` already does a comprehensive web assessment
@@ -1360,6 +1361,22 @@ pub async fn run(mut cfg: RunConfig, lib: &Library, pool: &ModelPool, tx: Sender
let shots = if evidence_dir.is_empty() { String::new() } else { screenshot_doctrine(&evidence_dir) };
// Token economy: each agent gets a capped recon context, not the full blob.
let recon_ctx: String = recon.chars().take(3500).collect();
// Resume support: skip agents that already ran in a previous invocation.
let snapshot = cfg.workdir.as_deref().and_then(crate::resilience::RunSnapshot::load);
let mut run_snap = if let Some(ref snap) = snapshot {
let _ = tx.send(format!("📂 resuming from snapshot: {} agents done, {} findings carried", snap.agents_completed.len(), snap.findings_so_far.len())).await;
snap.clone()
} else {
crate::resilience::RunSnapshot::new(&cfg.target, &recon_ctx)
};
run_snap.agents_pending = selected.iter().map(|a| a.name.clone()).collect();
run_snap.set_phase("exploit");
// Budget warnings: progressive alerts.
let budget_warner = std::sync::Arc::new(crate::resilience::BudgetWarner::new());
let snap_ref = snapshot.as_ref();
let raw: Vec<(String, String, Vec<Finding>)> = stream::iter(selected.iter().cloned())
.map(|ag| {
let target = target.clone();
@@ -1370,9 +1387,27 @@ pub async fn run(mut cfg: RunConfig, lib: &Library, pool: &ModelPool, tx: Sender
let txc = tx.clone();
let wd = cfg.workdir.clone();
async move {
// Resume: skip agents already completed in a prior run.
if let Some(snap) = snap_ref {
if snap.already_ran(&ag.name) {
let _ = txc.send(format!("⏩ skipping {} (already ran in prior run)", ag.name)).await;
let prior: Vec<Finding> = snap.findings_so_far.iter()
.filter(|f| f.agent == ag.name)
.cloned().collect();
return (ag.name.clone(), String::new(), prior);
}
}
if pool.stop_exploiting() {
return (ag.name.clone(), String::new(), vec![]);
}
// Budget gate: stop spawning new agents at 95%.
if crate::resilience::BudgetWarner::should_stop_new_agents(
pool.tokens_used(),
pool.token_limit(),
) {
let _ = txc.send(format!("🛑 budget exhausted — skipping agent {}", ag.name)).await;
return (ag.name.clone(), String::new(), vec![]);
}
if verbose {
let _ = txc.send(format!(" ▶ launching agent: {} ({})", ag.name, ag.title.replace(" Agent", ""))).await;
}
@@ -1445,6 +1480,15 @@ pub async fn run(mut cfg: RunConfig, lib: &Library, pool: &ModelPool, tx: Sender
} else {
let _ = txc.send(format!("exploit {} failed: {e}", ag.name)).await;
}
// Nudge: retry with simplified prompt (skip for auth failures).
if !is_auth && !pool.stop_exploiting() {
let _ = txc.send(format!("🔄 nudging {} with simplified prompt...", ag.name)).await;
let nudged = crate::resilience::nudge_agent(pool, &ag.name, &ag.system, &user, Task::Exploit).await;
if !nudged.is_empty() {
let _ = txc.send(format!("✓ nudge recovered {} finding(s) from {}", nudged.len(), ag.name)).await;
return (ag.name.clone(), String::new(), nudged);
}
}
let error_text = format!("ERROR: {e}");
let salvaged = crate::resilience::salvage_findings(&ag.name, &error_text, wd.as_deref());
if !salvaged.is_empty() {
@@ -1459,6 +1503,21 @@ pub async fn run(mut cfg: RunConfig, lib: &Library, pool: &ModelPool, tx: Sender
.collect()
.await;
// Update snapshot with completed agents.
for (name, _, findings) in &raw {
run_snap.mark_agent_done(name, findings);
}
run_snap.set_phase("validation");
if let Some(ref wd) = cfg.workdir {
let _ = run_snap.save(wd);
}
// Budget warning after exploit phase.
if let Some(threshold) = budget_warner.check(pool.tokens_used(), pool.token_limit()) {
let msg = crate::resilience::BudgetWarner::message(threshold, pool.tokens_used(), pool.token_limit(), 0);
let _ = tx.send(format!("notify: {msg}")).await;
}
// Merge any previously salvaged findings from prior crashed runs.
let mut prior_salvaged = if let Some(ref wd) = cfg.workdir {
crate::resilience::load_salvaged(wd)
+29 -1
View File
@@ -90,6 +90,10 @@ pub struct ModelPool {
/// Backends already tried as an automatic fallback, so a failing one is not
/// retried in a loop.
tried_auto: Arc<Mutex<Vec<String>>>,
/// Approximate tokens consumed (tracked by response length * 4 heuristic).
tokens_used: Arc<std::sync::atomic::AtomicU64>,
/// Token budget ceiling (0 = unlimited).
token_limit: std::sync::atomic::AtomicU64,
}
impl ModelPool {
@@ -124,6 +128,8 @@ impl ModelPool {
fallback: Arc::new(Mutex::new(Vec::new())),
consecutive_auth_fails: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
tried_auto: Arc::new(Mutex::new(Vec::new())),
tokens_used: Arc::new(std::sync::atomic::AtomicU64::new(0)),
token_limit: std::sync::atomic::AtomicU64::new(0),
}
}
@@ -167,6 +173,27 @@ impl ModelPool {
|| self.soft.load(std::sync::atomic::Ordering::Relaxed)
}
/// Set the token budget ceiling for progressive budget warnings.
pub fn set_token_limit(&self, limit: u64) {
self.token_limit.store(limit, std::sync::atomic::Ordering::Relaxed);
}
/// Approximate tokens consumed so far.
pub fn tokens_used(&self) -> u64 {
self.tokens_used.load(std::sync::atomic::Ordering::Relaxed)
}
/// Token budget ceiling (0 = unlimited).
pub fn token_limit(&self) -> u64 {
self.token_limit.load(std::sync::atomic::Ordering::Relaxed)
}
/// Record approximate token usage from a response.
pub fn record_tokens(&self, response_chars: usize) {
let approx = (response_chars / 4) as u64;
self.tokens_used.fetch_add(approx, std::sync::atomic::Ordering::Relaxed);
}
/// Handle to the PAUSE flag (observe whether the run is parked on exhaustion).
pub fn pause_handle(&self) -> Arc<AtomicBool> {
self.paused.clone()
@@ -385,7 +412,8 @@ impl ModelPool {
}
match self.one(label, m, system, user).await {
Ok(text) => {
self.reset_auth_fails(); // success resets circuit breaker
self.reset_auth_fails();
self.record_tokens(text.len());
return Ok((m.clone(), text));
}
Err(e) => {
+432 -4
View File
@@ -249,9 +249,7 @@ pub async fn maybe_compact_prompt(
);
Some(compacted)
}
Err(e) => {
None
}
Err(_) => None,
}
}
@@ -311,6 +309,349 @@ pub fn load_salvaged(workdir: &str) -> Vec<Finding> {
.unwrap_or_default()
}
// ---------------------------------------------------------------------------
// 5. LLM-based finding dedup (semantic, not just hash)
// ---------------------------------------------------------------------------
const DEDUP_SYSTEM: &str = "\
You are a vulnerability dedup judge. Given a CANDIDATE finding and EXISTING findings, \
decide if the candidate is a duplicate.\n\n\
SAME vulnerability means: same root cause, same affected endpoint/file, same exploitation \
method, would be fixed by the same code change.\n\n\
NOT duplicates if: different endpoints even with same vuln type, different parameters in \
same endpoint, different root causes (stored XSS vs reflected XSS), different auth context.\n\n\
ARE duplicates even if: titles differ, descriptions vary in detail, PoC uses different \
payloads but exploits same issue, one report is more thorough.\n\n\
Respond with ONLY a JSON object:\n\
{\"is_duplicate\": true/false, \"match_index\": <0-based index or -1>, \"confidence\": 0.0-1.0, \"reason\": \"...\"}\n\
No prose, no code fences.";
/// Check if a candidate finding duplicates any existing finding using LLM judgment.
/// Returns `Some(index)` of the matched existing finding, or `None` if unique.
pub async fn llm_dedup_check(
pool: &ModelPool,
candidate: &Finding,
existing: &[Finding],
) -> Option<usize> {
if existing.is_empty() {
return None;
}
// Fast-path: exact title+endpoint match (deterministic, no LLM call).
for (i, ex) in existing.iter().enumerate() {
if ex.title == candidate.title && ex.endpoint == candidate.endpoint {
return Some(i);
}
}
// Fast-path: same CWE + same endpoint key = very likely duplicate.
let cand_cwe = cwe_number(&candidate.cwe);
if cand_cwe != 0 {
for (i, ex) in existing.iter().enumerate() {
if cwe_number(&ex.cwe) == cand_cwe
&& normalize_endpoint(&ex.endpoint) == normalize_endpoint(&candidate.endpoint)
{
return Some(i);
}
}
}
// LLM dedup for fuzzy cases: different wording, different CWE but same root cause.
let existing_summary: Vec<String> = existing
.iter()
.enumerate()
.take(30) // cap to avoid context overflow
.map(|(i, f)| format!("[{i}] {} | {} | {} | CWE-{}", f.title, f.severity, f.endpoint, f.cwe))
.collect();
let user_msg = format!(
"CANDIDATE:\ntitle: {}\nseverity: {}\nendpoint: {}\ncwe: {}\nevidence: {}\n\nEXISTING:\n{}\n\nRespond with ONLY the JSON object.",
candidate.title, candidate.severity, candidate.endpoint, candidate.cwe,
&candidate.evidence[..candidate.evidence.len().min(500)],
existing_summary.join("\n")
);
match pool.complete(DEDUP_SYSTEM, &user_msg).await {
Ok((_model, response)) => parse_dedup_response(&response),
Err(_) => None, // LLM failed, treat as unique (safe default)
}
}
fn parse_dedup_response(text: &str) -> Option<usize> {
let start = text.find('{')?;
let end = text.rfind('}')?;
if end <= start { return None; }
let json: serde_json::Value = serde_json::from_str(&text[start..=end]).ok()?;
if json.get("is_duplicate")?.as_bool()? {
let idx = json.get("match_index")?.as_i64()?;
if idx >= 0 { Some(idx as usize) } else { None }
} else {
None
}
}
fn cwe_number(cwe: &str) -> u32 {
cwe.trim_start_matches("CWE-")
.trim_start_matches("cwe-")
.parse()
.unwrap_or(0)
}
fn normalize_endpoint(ep: &str) -> String {
ep.split(['?', '#']).next().unwrap_or(ep)
.trim_end_matches('/')
.to_lowercase()
}
// ---------------------------------------------------------------------------
// 6. Budget warnings (progressive alerts at 70/85/95%)
// ---------------------------------------------------------------------------
use std::sync::atomic::{AtomicU8, Ordering};
/// Thresholds at which to warn the operator about budget usage.
const WARN_THRESHOLDS: &[u8] = &[70, 85, 95];
/// Tracks which budget warning levels have already been emitted.
pub struct BudgetWarner {
last_warned: AtomicU8,
}
impl BudgetWarner {
pub fn new() -> Self {
Self { last_warned: AtomicU8::new(0) }
}
/// Check if a warning should fire. Returns `Some(pct)` with the threshold
/// crossed, or `None` if already warned at this level.
pub fn check(&self, spent: u64, limit: u64) -> Option<u8> {
if limit == 0 {
return None; // unlimited budget
}
let pct = ((spent as f64 / limit as f64) * 100.0) as u8;
let last = self.last_warned.load(Ordering::Relaxed);
for &threshold in WARN_THRESHOLDS {
if pct >= threshold && last < threshold {
self.last_warned.store(threshold, Ordering::Relaxed);
return Some(threshold);
}
}
None
}
/// Format a warning message for the given threshold.
pub fn message(threshold: u8, spent: u64, limit: u64, remaining_agents: usize) -> String {
let pct = if limit > 0 { (spent as f64 / limit as f64) * 100.0 } else { 0.0 };
match threshold {
70 => format!(
"⚠ budget {pct:.0}% used ({spent}/{limit} tokens). {} agents still running. Consider: reduce concurrency or switch to cheaper model.",
remaining_agents
),
85 => format!(
"⚠⚠ budget {pct:.0}% used ({spent}/{limit} tokens). Entering conservation mode — new agents use cheap analysis only."
),
95 => format!(
"🛑 budget {pct:.0}% used ({spent}/{limit} tokens). Reserving remaining tokens for validation of findings already collected. No new exploit agents."
),
_ => format!("⚠ budget {pct:.0}% used ({spent}/{limit} tokens)"),
}
}
/// At 90%+ budget, should we reserve tokens for sub-agent work (validation)?
pub fn should_reserve_for_validation(spent: u64, limit: u64) -> bool {
if limit == 0 { return false; }
let pct = (spent as f64 / limit as f64) * 100.0;
pct >= 90.0
}
/// At 95%+, should we stop spawning new exploit agents?
pub fn should_stop_new_agents(spent: u64, limit: u64) -> bool {
if limit == 0 { return false; }
let pct = (spent as f64 / limit as f64) * 100.0;
pct >= 95.0
}
}
// ---------------------------------------------------------------------------
// 7. Agent lifecycle nudge (retry failed agents with simplified prompt)
// ---------------------------------------------------------------------------
const MAX_NUDGE_ATTEMPTS: usize = 3;
const NUDGE_PREFIX: &str = "\
Your previous attempt failed or produced no results. This is a RETRY with a \
simplified prompt. Focus on the MOST LIKELY vulnerability for your speciality. \
Be direct, test one thing, and report what you find.\n\n";
/// Retry a failed single-shot agent with progressively simplified prompts.
/// Returns findings from the first successful retry, or empty if all fail.
pub async fn nudge_agent(
pool: &ModelPool,
agent_name: &str,
system: &str,
original_user: &str,
task: crate::pool::Task,
) -> Vec<Finding> {
for attempt in 1..=MAX_NUDGE_ATTEMPTS {
let simplified = match attempt {
1 => format!("{NUDGE_PREFIX}{original_user}"),
2 => {
// Trim recon context to reduce noise.
let trimmed: String = original_user.chars().take(2000).collect();
format!("{NUDGE_PREFIX}SIMPLIFIED:\n{trimmed}")
}
_ => {
// Final attempt: bare minimum prompt.
let core: String = original_user.lines()
.filter(|l| l.contains("target") || l.contains("AUTHORIZED") || l.contains("Reply ONLY"))
.collect::<Vec<_>>()
.join("\n");
format!("{NUDGE_PREFIX}LAST ATTEMPT — minimal context:\n{core}")
}
};
match pool.complete_routed(task, agent_name, system, &simplified).await {
Ok((_model, text)) => {
let findings = crate::pipeline::extract_findings(&text, agent_name);
if !findings.is_empty() {
return findings;
}
// Got a response but no findings — try next simplification.
}
Err(_) => {
// Complete failure — try next simplification.
tokio::time::sleep(Duration::from_secs(2)).await;
}
}
}
Vec::new()
}
// ---------------------------------------------------------------------------
// 8. Image budget enforcement (cap screenshots in context)
// ---------------------------------------------------------------------------
/// Strip base64 image data from a prompt to reduce token usage.
/// Keeps the first `max_images` image references and replaces the rest
/// with `[image elided]`.
pub fn enforce_image_budget(text: &str, max_images: usize) -> String {
let mut result = String::with_capacity(text.len());
let mut images_kept = 0;
let mut remaining = text;
while let Some(start) = remaining.find("data:image/") {
result.push_str(&remaining[..start]);
// Find the end of the base64 data.
let after = &remaining[start..];
let end = after.find('"')
.or_else(|| after.find('\''))
.or_else(|| after.find('\n'))
.or_else(|| after.find(' '))
.unwrap_or(after.len());
if images_kept < max_images {
result.push_str(&remaining[start..start + end]);
images_kept += 1;
} else {
result.push_str("[screenshot elided to bound context]");
}
remaining = &remaining[start + end..];
}
result.push_str(remaining);
result
}
/// Strip all base64 images from text (useful for image-rejection recovery).
pub fn strip_all_images(text: &str) -> String {
enforce_image_budget(text, 0)
}
// ---------------------------------------------------------------------------
// 9. Run snapshot/resume (persist state for interrupted runs)
// ---------------------------------------------------------------------------
use serde::{Deserialize, Serialize};
/// Snapshot of an in-progress run, serializable to JSON.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunSnapshot {
pub target: String,
pub recon: String,
pub agents_completed: Vec<String>,
pub agents_pending: Vec<String>,
pub findings_so_far: Vec<Finding>,
pub phase: String,
pub timestamp: String,
}
impl RunSnapshot {
pub fn new(target: &str, recon: &str) -> Self {
Self {
target: target.to_string(),
recon: recon.to_string(),
agents_completed: Vec::new(),
agents_pending: Vec::new(),
findings_so_far: Vec::new(),
phase: "init".to_string(),
timestamp: chrono_now(),
}
}
pub fn mark_agent_done(&mut self, name: &str, findings: &[Finding]) {
self.agents_pending.retain(|n| n != name);
if !self.agents_completed.contains(&name.to_string()) {
self.agents_completed.push(name.to_string());
}
for f in findings {
if !self.findings_so_far.iter().any(|e| e.title == f.title && e.endpoint == f.endpoint) {
self.findings_so_far.push(f.clone());
}
}
self.timestamp = chrono_now();
}
pub fn set_phase(&mut self, phase: &str) {
self.phase = phase.to_string();
self.timestamp = chrono_now();
}
/// Persist snapshot to the run's workdir.
pub fn save(&self, workdir: &str) -> std::io::Result<()> {
let path = Path::new(workdir).join("run_snapshot.json");
let json = serde_json::to_string_pretty(self)
.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
std::fs::write(path, json)
}
/// Load a previous snapshot from a workdir.
pub fn load(workdir: &str) -> Option<Self> {
let path = Path::new(workdir).join("run_snapshot.json");
let data = std::fs::read_to_string(path).ok()?;
serde_json::from_str(&data).ok()
}
/// Agents that haven't completed yet (for resume).
pub fn remaining_agents(&self) -> &[String] {
&self.agents_pending
}
/// Has this agent already been run in a prior invocation?
pub fn already_ran(&self, agent_name: &str) -> bool {
self.agents_completed.iter().any(|n| n == agent_name)
}
}
fn chrono_now() -> String {
use std::time::SystemTime;
let d = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.unwrap_or_default();
format!("{}", d.as_secs())
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
@@ -401,7 +742,94 @@ mod tests {
#[test]
fn compact_threshold_is_sane() {
// 100k tokens * 4 chars = 400k chars before compaction triggers.
assert!(COMPACT_THRESHOLD_TOKENS > KEEP_RECENT_TOKENS);
}
// --- Dedup helpers ---
#[test]
fn cwe_number_parses() {
assert_eq!(cwe_number("CWE-79"), 79);
assert_eq!(cwe_number("cwe-89"), 89);
assert_eq!(cwe_number("89"), 89);
assert_eq!(cwe_number(""), 0);
}
#[test]
fn normalize_endpoint_strips_query() {
assert_eq!(normalize_endpoint("/api/users?id=1"), "/api/users");
assert_eq!(normalize_endpoint("/API/Users/"), "/api/users");
}
// --- Budget warner ---
#[test]
fn budget_warner_fires_at_thresholds() {
let w = BudgetWarner::new();
assert_eq!(w.check(50, 100), None); // 50% — no warning
assert_eq!(w.check(70, 100), Some(70)); // 70% — first warning
assert_eq!(w.check(72, 100), None); // 72% — already warned at 70
assert_eq!(w.check(85, 100), Some(85)); // 85% — second warning
assert_eq!(w.check(95, 100), Some(95)); // 95% — final warning
assert_eq!(w.check(99, 100), None); // 99% — already warned at 95
}
#[test]
fn budget_warner_unlimited_never_fires() {
let w = BudgetWarner::new();
assert_eq!(w.check(999999, 0), None);
}
#[test]
fn budget_reserve_triggers_at_90() {
assert!(!BudgetWarner::should_reserve_for_validation(80, 100));
assert!(BudgetWarner::should_reserve_for_validation(91, 100));
assert!(!BudgetWarner::should_reserve_for_validation(91, 0)); // unlimited
}
#[test]
fn budget_stop_agents_at_95() {
assert!(!BudgetWarner::should_stop_new_agents(94, 100));
assert!(BudgetWarner::should_stop_new_agents(96, 100));
}
// --- Image budget ---
#[test]
fn image_budget_keeps_n_images() {
let text = r#"before data:image/png;base64,AAA" middle data:image/jpg;base64,BBB" end"#;
let result = enforce_image_budget(text, 1);
assert!(result.contains("data:image/png;base64,AAA"));
assert!(result.contains("[screenshot elided"));
assert!(!result.contains("BBB"));
}
#[test]
fn strip_all_images_removes_all() {
let text = r#"img data:image/png;base64,AAAA" done"#;
let result = strip_all_images(text);
assert!(!result.contains("AAAA"));
assert!(result.contains("[screenshot elided"));
}
// --- Run snapshot ---
#[test]
fn snapshot_tracks_agents() {
let mut snap = RunSnapshot::new("https://target.test", "{}");
snap.agents_pending = vec!["sqli".into(), "xss".into(), "csrf".into()];
assert!(!snap.already_ran("sqli"));
snap.mark_agent_done("sqli", &[]);
assert!(snap.already_ran("sqli"));
assert_eq!(snap.agents_pending.len(), 2);
assert_eq!(snap.agents_completed.len(), 1);
}
#[test]
fn dedup_response_parsing() {
let good = r#"{"is_duplicate": true, "match_index": 2, "confidence": 0.95, "reason": "same vuln"}"#;
assert_eq!(parse_dedup_response(good), Some(2));
let not_dup = r#"{"is_duplicate": false, "match_index": -1, "confidence": 0.9, "reason": "different"}"#;
assert_eq!(parse_dedup_response(not_dup), None);
let garbage = "not json at all";
assert_eq!(parse_dedup_response(garbage), None);
}
}