bruteforce: cap per-host concurrency in batch, wire combo-file (-C), expose delay/jitter, medusa host give-up

1. Batch concurrency cap: in a mass-scan fan-out, per-host bruteforce concurrency
   is capped (4) so it no longer multiplies with scheduler host-concurrency into
   an RLIMIT_NOFILE-exhausting socket count.
2. Combo-file (hydra -C): 'setg credential_file <user:pass file>' is loaded via
   the engine's load_credential_file and tried alongside defaults. Verified:
   2-pair file -> +2 attempts.
3. Delay/jitter (hydra -w): 'setg bruteforce_delay_ms' + 'setg bruteforce_jitter_ms'
   now feed BruteforceConfig.delay_ms/jitter_ms (were hardcoded 0).
4. Host give-up (medusa): run_bruteforce now aborts a host after MAX_LOCKOUT_PAUSES
   (3) consecutive-error lockout pauses with no success, instead of pausing-and-
   grinding forever. run_subnet_bruteforce already had give-up.

Also surfaced credential_file/bruteforce_delay_ms/bruteforce_jitter_ms in
'show options'. Build: 0 errors, 0 warnings.
Known limitation: streaming-path give-up is per-batch (not cross-batch).

src/shell.rs

@@ -1311,6 +1311,9 @@ async fn display_all_options(ctx: &ShellContext) {
         ("username_wordlist",  "Bruteforce username wordlist path", ""),
         ("password_wordlist",  "Bruteforce password wordlist path", ""),
         ("cred_extras",        "Bruteforce extra creds: hydra-style n/s/r (null/same/reversed), e.g. nsr", ""),
+        ("credential_file",    "Bruteforce user:pass combo file (hydra -C)", ""),
+        ("bruteforce_delay_ms","Per-attempt delay to dodge lockout (ms, hydra -w)", ""),
+        ("bruteforce_jitter_ms","Random extra delay added to bruteforce_delay_ms (ms)", ""),
     ];
 
     let mut displayed = std::collections::HashSet::new();

src/utils/bruteforce.rs

@@ -489,6 +489,10 @@ pub enum ComboMode {
 /// `run_bruteforce_streaming` so we don't OOM on a 10⁷×10⁵ wordlist combo.
 const MAX_COMBOS: usize = 10_000_000;
 
+/// Number of 10-consecutive-error lockout pauses tolerated (with no success)
+/// before the engine gives up on a host instead of pausing-and-grinding forever.
+const MAX_LOCKOUT_PAUSES: u64 = 3;
+
 /// Generate credential pairs with full combo mode control.
 /// - Linear: pairs user\[i\] with pass\[i\], cycling the shorter list.
 /// - Combo: full cross product (user × pass).
@@ -727,6 +731,11 @@ where
     let errors: BruteforceErrorList = Arc::new(Mutex::new(Vec::new()));
     let stop = Arc::new(AtomicBool::new(false));
     let paused = Arc::new(AtomicBool::new(false));
+    // Medusa-style "give up on this host": set when repeated lockout pauses make
+    // no progress. Unlike `stop` (which is gated on stop_on_success), `abort`
+    // always halts the run so a dead/blocking host can't grind the whole wordlist.
+    let abort = Arc::new(AtomicBool::new(false));
+    let lockout_pauses = Arc::new(AtomicU64::new(0));
     let semaphore = Arc::new(Semaphore::new(config.concurrency.max(1)));
 
     // Progress reporter
@@ -745,6 +754,9 @@ where
     let mut tasks = FuturesUnordered::new();
 
     for (user, pass) in combos {
+        if abort.load(Ordering::Relaxed) {
+            break;
+        }
         if config.stop_on_success && stop.load(Ordering::Relaxed) {
             break;
         }
@@ -766,6 +778,8 @@ where
         let errors_c = errors.clone();
         let stop_c = stop.clone();
         let paused_c = paused.clone();
+        let abort_c = abort.clone();
+        let lockout_pauses_c = lockout_pauses.clone();
         let stats_c = stats.clone();
         let try_login_c = try_login.clone();
         let verbose = config.verbose;
@@ -783,6 +797,7 @@ where
         tasks.push(tokio::spawn(async move {
             let _permit = permit;
             let __body = async move {
+            if abort_c.load(Ordering::Relaxed) { return; }
             if stop_on_success && stop_c.load(Ordering::Relaxed) { return; }
 
             // Respect global rate-limit pause
@@ -833,7 +848,29 @@ where
                         }
                         stats_c.record_error(message.clone()).await;
                         if stats_c.is_lockout_likely(10) && !paused_c.swap(true, Ordering::AcqRel) {
-                            crate::meprintln!("\n{}", "[!] WARNING: 10+ consecutive errors — possible rate limiting. Pausing 30s...".red().bold());
+                            let pauses = lockout_pauses_c.fetch_add(1, Ordering::Relaxed) + 1;
+                            // Give up (medusa-style) once repeated pauses make no
+                            // progress: a dead host / wrong protocol / hard block
+                            // shouldn't pause-and-grind the whole wordlist forever.
+                            if pauses >= MAX_LOCKOUT_PAUSES && found_c.lock().await.is_empty() {
+                                crate::meprintln!(
+                                    "\n{}",
+                                    format!(
+                                        "[!] Giving up on {} — {} lockout pauses, no success (host blocking/down?)",
+                                        display, pauses
+                                    ).red().bold()
+                                );
+                                abort_c.store(true, Ordering::Relaxed);
+                                paused_c.store(false, Ordering::Release);
+                                break;
+                            }
+                            crate::meprintln!(
+                                "\n{}",
+                                format!(
+                                    "[!] WARNING: 10+ consecutive errors — possible rate limiting. Pausing 30s... (pause {}/{})",
+                                    pauses, MAX_LOCKOUT_PAUSES
+                                ).red().bold()
+                            );
                             tokio::time::sleep(Duration::from_secs(30)).await;
                             stats_c.consecutive_errors.store(0, Ordering::Relaxed);
                             paused_c.store(false, Ordering::Release);

src/utils/creds_helper.rs

@@ -18,11 +18,16 @@ use colored::*;
 use crate::module::{Finding, FindingKind, ModuleOutcome};
 use crate::utils::{
     cfg_prompt_default, cfg_prompt_existing_file, cfg_prompt_int_range,
-    cfg_prompt_yes_no, file_size, generate_combos_mode, load_lines, normalize_target,
-    parse_combo_mode, run_bruteforce, run_bruteforce_streaming, BruteforceConfig,
-    LoginResult, STREAMING_THRESHOLD,
+    cfg_prompt_yes_no, file_size, generate_combos_mode, load_credential_file, load_lines,
+    normalize_target, parse_combo_mode, run_bruteforce, run_bruteforce_streaming,
+    BruteforceConfig, LoginResult, STREAMING_THRESHOLD,
 };
 
+/// Per-host bruteforce concurrency cap when running inside a mass-scan fan-out.
+/// The scheduler already parallelises across hosts, so this keeps the total open
+/// socket count (scheduler_hosts x this) well under typical RLIMIT_NOFILE.
+const BATCH_PER_HOST_CONCURRENCY: usize = 4;
+
 /// `read_exact` with a wall-clock timeout, flattened to a single
 /// `io::Result<()>`. Avoids the nested-match `Ok(Ok(_)) / Ok(Err) / Err(_)`
 /// pattern that's easy to miswrite.
@@ -136,8 +141,15 @@ where
     )
     .await?;
     let mode = parse_combo_mode(&combo_input);
-    let concurrency =
+    let mut concurrency =
         cfg_prompt_int_range("concurrency", "Concurrency", 16, 1, 256).await? as usize;
+    // In a mass-scan fan-out the scheduler already runs many hosts concurrently,
+    // so a high PER-HOST bruteforce concurrency multiplies into the total open
+    // socket count (e.g. 50 hosts x 16 = 800) and can blow past RLIMIT_NOFILE.
+    // Cap per-host concurrency in batch mode; the scheduler supplies the breadth.
+    if crate::utils::is_batch_mode() {
+        concurrency = concurrency.min(BATCH_PER_HOST_CONCURRENCY);
+    }
     let timeout_secs =
         cfg_prompt_int_range("timeout", "Per-attempt timeout (seconds)", 5, 1, 60).await? as u64;
     let stop_on_success =
@@ -215,14 +227,36 @@ where
         }
     }
 
+    // Hydra `-C`: a colon-separated user:pass credential file, if provided via
+    // `setg credential_file`, is loaded and tried alongside the defaults.
+    if let Some(cred_file) = opt_str("credential_file") {
+        match load_credential_file(&cred_file) {
+            Ok(pairs) => {
+                tracing::debug!(
+                    "{}: loaded {} credential pair(s) from {}",
+                    cfg.service_name,
+                    pairs.len(),
+                    cred_file
+                );
+                defaults.extend(pairs);
+            }
+            Err(e) => return Err(anyhow!("credential_file '{}': {}", cred_file, e)),
+        }
+    }
+
+    // Hydra `-w`-style throttle to dodge account lockout: `setg bruteforce_delay_ms`
+    // (fixed delay after each attempt) + `setg bruteforce_jitter_ms` (random extra).
+    let delay_ms = opt_u64("bruteforce_delay_ms");
+    let jitter_ms = opt_u64("bruteforce_jitter_ms");
+
     let bf_config = BruteforceConfig {
         target: host.clone(),
         port,
         concurrency,
         stop_on_success,
         verbose: false,
-        delay_ms: 0,
-        jitter_ms: 0,
+        delay_ms,
+        jitter_ms,
         max_retries: 1,
         service_name: cfg.service_name,
         source_module: cfg.source_module,
@@ -396,3 +430,27 @@ fn cred_extras() -> CredExtras {
         reversed: raw.contains('r'),
     }
 }
+
+/// Read a non-empty global option (`setg <key>`), or `None` if unset/blank.
+fn opt_str(key: &str) -> Option<String> {
+    crate::tenant::resolve()
+        .global_options()
+        .try_get(key)
+        .filter(|s| !s.trim().is_empty())
+}
+
+/// Read a `u64` global option (`setg <key>`), defaulting to 0 when unset or
+/// unparseable (the error is surfaced at debug, not silently dropped).
+fn opt_u64(key: &str) -> u64 {
+    let raw = match crate::tenant::resolve().global_options().try_get(key) {
+        Some(v) => v,
+        None => return 0,
+    };
+    match raw.trim().parse::<u64>() {
+        Ok(n) => n,
+        Err(e) => {
+            tracing::debug!("{key} is not a valid u64 ('{}'): {e}; using 0", raw.trim());
+            0
+        }
+    }
+}