fix: friends whole-tool review — 15 fabrication/silent-drop/coverage fixes

auth_utils.py

@@ -13,6 +13,7 @@
 import json
 import os
 import struct
+import sys
 import tempfile
 import threading
 import time
@@ -258,6 +259,13 @@ def __init__(self, base_url: str, rate_limiter: RateLimiter = None):
         self._creds = None
         self._login_url = None
         self.token = None
+        # friends full-tool review F14: count transport failures (TLS/timeout/conn
+        # reset) separately from real HTTP responses. request() used to collapse
+        # every exception to status:0, which downstream 'status in (200,201)' checks
+        # treat as a benign non-hit — so a systematic transport failure produced a
+        # clean-looking 0-finding run with no coverage-loss signal.
+        self.transport_errors = 0
+        self.total_requests = 0
         # Cookie name carrying the auth token for explicit-token requests. Defaults to
         # 'cf_at' but is overwritten in auto_login() with whichever cookie the server
         # actually set the JWT in (cf_at/access_token/jwt/token/session), so cookie-only
@@ -568,6 +576,7 @@ def request(self, method: str, path: str, token: str = None,
                 body = resp.json()
             except Exception:
                 body = resp.text[:2000]
+            self.total_requests += 1
             return {
                 "status": resp.status_code,
                 "headers": dict(resp.headers),
@@ -576,7 +585,17 @@ def request(self, method: str, path: str, token: str = None,
                 "method": method,
             }
         except Exception as e:
-            return {"status": 0, "headers": {}, "body": str(e), "url": url, "method": method}
+            # F14: a transport failure is NOT an absent endpoint. Mark it distinctly
+            # and count it so a systematic failure (bad token, TLS, proxy outage)
+            # surfaces as lost coverage instead of a benign 0-finding run.
+            self.total_requests += 1
+            self.transport_errors += 1
+            if self.transport_errors in (1, 5, 25) or self.transport_errors % 100 == 0:
+                print(f"[auth_utils] WARNING: transport failure #{self.transport_errors} "
+                      f"({type(e).__name__}) on {method} {url} — these are COVERAGE LOSS, "
+                      f"not clean non-hits: {str(e)[:120]}", file=sys.stderr)
+            return {"status": 0, "headers": {}, "body": str(e), "url": url,
+                    "method": method, "transport_error": True, "error": str(e)}
 
 
 class FindingSaver:

autopilot_api_hunt.py

@@ -770,6 +770,13 @@ def run(self, session: AuthSession, endpoints: list[dict],
                         f"were not tested for IDOR."))
         for ep in idor_targets[:_IDOR_CAP]:
             path = ep["path"]
+            # F11 owner-baseline: a single 200-with-PII is NOT proof of IDOR — the
+            # endpoint may ignore the id and return the CALLER's OWN record for every
+            # value (correct behaviour). Real IDOR requires that DIFFERENT ids return
+            # DIFFERENT records. Collect the PII value-set per (request shape, id) and
+            # only confirm when >=2 distinct non-empty PII sets appear across >=2 ids
+            # for the SAME shape. Emit once per endpoint, not once per id.
+            shape_hits: dict = {}   # shape_key -> {test_id: (exposed, sample, url, signature)}
             for test_id in [1, 2, 3, 100]:
                 # Try both FormData and JSON (Django APIs typically use FormData)
                 for payload in [
@@ -778,6 +785,8 @@ def run(self, session: AuthSession, endpoints: list[dict],
                     {"data": {"id": str(test_id), "course_id": str(test_id)}},
                     {"data": {"learner_id": str(test_id)}},
                 ]:
+                    shape_key = tuple(sorted(payload)) + tuple(
+                        sorted(k for v in payload.values() if isinstance(v, dict) for k in v))
                     resp = session.request("POST", path, **payload)
                     if resp["status"] in (200, 201) and isinstance(resp["body"], dict):
                         body = resp["body"]
@@ -790,21 +799,35 @@ def run(self, session: AuthSession, endpoints: list[dict],
                             if not isinstance(data, dict):
                                 continue
                             exposed = {k for k in data.keys() if k.lower() in self.PII_KEYS}
-                            if exposed:
-                                sample = ', '.join(f'{k}={data[k]}' for k in list(exposed)[:3] if data.get(k))
-                                f = {"type": "idor", "severity": HIGH,
-                                     "detail": f"IDOR: id={test_id} exposes PII ({', '.join(exposed)}) on {path}",
-                                     "url": resp["url"],
-                                     "evidence": f"id={test_id} → {sample}"}
-                                findings.append(f)
-                                if saver:
-                                    saver.save(f)
-                                    saver.save_txt(f)
-                                break
-                        if findings and findings[-1]["detail"].startswith(f"IDOR: id={test_id}"):
-                            break  # Found IDOR with this payload format
-                if any(f["detail"].startswith(f"IDOR: id={test_id}") for f in findings):
-                    break  # One confirmed IDOR per endpoint
+                            if not exposed:
+                                continue
+                            signature = frozenset((k, str(data.get(k))) for k in exposed)
+                            sample = ', '.join(f'{k}={data[k]}' for k in list(exposed)[:3] if data.get(k))
+                            shape_hits.setdefault(shape_key, {})[test_id] = (
+                                exposed, sample, resp["url"], signature)
+            # Decide per request shape: IDOR only when >=2 ids returned >=2 DISTINCT
+            # PII records. Identical PII across ids == the caller's own record echoed
+            # (benign) -> no finding. Emit at most ONE finding per endpoint (the first
+            # confirming shape) rather than one per payload variant.
+            for hits in shape_hits.values():
+                distinct_sigs = {h[3] for h in hits.values() if h[3]}
+                if len(hits) < 2 or len(distinct_sigs) < 2:
+                    continue
+                reps = sorted(hits.items())
+                (id1, h1), (id2, h2) = reps[0], reps[1]
+                all_exposed = sorted(set().union(*[h[0] for h in hits.values()]))
+                f = {"type": "idor", "severity": HIGH,
+                     "detail": (f"IDOR on {path}: different ids return different records "
+                                f"(PII: {', '.join(all_exposed)})"),
+                     "url": h1[2],
+                     "evidence": (f"id={id1} → {h1[1]} | id={id2} → {h2[1]} "
+                                  f"— distinct records confirm the id selects another "
+                                  f"user's data")}
+                findings.append(f)
+                if saver:
+                    saver.save(f)
+                    saver.save_txt(f)
+                break  # one confirmed IDOR per endpoint is enough
 
         log("ok", f"  {len(findings)} IDOR findings")
         return findings
@@ -942,18 +965,14 @@ def run(self, session: AuthSession, saver: FindingSaver = None) -> list[dict]:
                 if saver:
                     saver.save(f)
                     saver.save_txt(f)
-                # Extract AWS key
-                url = resp["body"]["uploadUrl"]
-                key_match = re.search(r"Credential=([A-Z0-9]+)%2F", url)
-                bucket_match = re.search(r"https://([^/]+)\.s3\.", url)
-                if key_match:
-                    f2 = {"type": "aws_key_exposed", "severity": HIGH,
-                          "detail": f"AWS Access Key ID exposed: {key_match.group(1)}",
-                          "url": url[:80], "evidence": f"Key in presigned URL"}
-                    findings.append(f2)
-                    if saver:
-                        saver.save(f2)
-                        saver.save_txt(f2)
+                # NOTE (friends full-tool review F12): do NOT report the
+                # ``X-Amz-Credential=AKIA...`` value from the presigned URL as a
+                # credential leak. The AWS Access Key *ID* is a PUBLIC identifier
+                # present in EVERY SigV4 signature by design; the secret key signs
+                # the URL but never appears in it. Flagging the key ID produced a
+                # HIGH ``aws_key_exposed`` false positive on any working S3
+                # presigned-upload API. The real signal (a presigned URL minted for
+                # a dangerous filename/type) is already captured as ``f`` above.
                 break  # One confirmed is enough for S3
 
         # Test profile image upload bypass

brain_scanner.py

@@ -441,22 +441,70 @@ def _is_usage_banner(stdout: str) -> bool:
     return True
 
 
-def _verdict_findings(response: str, grounded: bool) -> list:
+# friends full-tool review F9: signals that a grounded stdout actually PROVES
+# exploitation (command output, extracted data, SQL/EL errors, file content,
+# reflected markers) — not merely a passive server banner/version string.
+_EXPLOIT_EVIDENCE_RE = re.compile(
+    r"uid=\d|gid=\d|groups=\d|root:.{0,3}:0:0:|/bin/(ba|z)?sh\b|"      # command exec / /etc/passwd
+    r"SQL syntax|ORA-\d{4,5}|SQLSTATE|PSQLException|"                  # SQL errors
+    r"syntax error at or near|Unclosed quotation|mysql_fetch|"        # SQL errors
+    r"-----BEGIN [A-Z]|<\?php|BEGIN RSA|ssh-rsa AAAA|"                 # file/key content
+    r"HTTP/\d\.\d 500|Traceback \(most recent call last\)|"           # server error / stack
+    r"onerror=|<script>alert|alert\(document|"                        # XSS reflection proof
+    r"\bwin\.ini\b|\[boot loader\]|Directory of ",                    # windows file read
+    re.I)
+# A line that is ONLY a server/version fingerprint (banner) carries no proof.
+_PASSIVE_FINGERPRINT_RE = re.compile(
+    r"^\s*(server\s*:|x-powered-by\s*:|via\s*:|x-aspnet(-mvc)?-version\s*:|"
+    r"x-generator\s*:|set-cookie\s*:|date\s*:|content-type\s*:|"
+    r"[\w./+-]+/\d+(\.\d+)+\s*)$", re.I)
+
+
+def _grounded_output_is_passive_only(text: str) -> bool:
+    """True when the grounded stdout is JUST passive fingerprint/banner data (or
+    empty) — no exploitation evidence — so it cannot corroborate a CONFIRMED
+    verdict (F9). Conservative: if ANY line is substantive-but-unclassified we
+    return False (do NOT downgrade), so a real exploit whose evidence shape we
+    don't recognise is never dropped."""
+    s = (text or "").strip()
+    if not s:
+        return True                              # no evidence at all
+    if _EXPLOIT_EVIDENCE_RE.search(s):
+        return False                             # real exploitation evidence present
+    for line in s.splitlines():
+        ln = line.strip()
+        if not ln:
+            continue
+        # A non-empty line that is neither a recognised banner nor trivially short
+        # is "substantive content we can't classify" → keep [VERIFIED], don't regress.
+        if not _PASSIVE_FINGERPRINT_RE.match(ln) and len(ln) > 24:
+            return False
+    return True                                  # everything was banner/short/trivial
+
+
+def _verdict_findings(response: str, grounded: bool, grounded_output: str = None) -> list:
     """Findings to record from an ACCEPTED final verdict, so a confirmed result is captured
     in the report (was silently lost as 'Findings: 0' when the severity word and 'CONFIRMED'
     landed on SEPARATE lines).
 
-    A GROUNDED verdict (>=1 script produced real output — guaranteed by the gate that lets a
-    CONFIRMED verdict through) is tagged ``[VERIFIED ...]`` so reporter verification-gating
-    KEEPS it; an ungrounded one stays ``[MODEL CLAIM ...]`` (the reporter drops those at med+).
+    A GROUNDED verdict (>=1 script produced real output) is tagged ``[VERIFIED ...]`` so
+    reporter verification-gating KEEPS it; an ungrounded one stays ``[MODEL CLAIM ...]``
+    (the reporter drops those at med+). friends full-tool review F9: grounding is only
+    trustworthy if the grounded stdout actually corroborates the claim — a passive server
+    banner grounds nothing. When ``grounded_output`` is supplied and is purely passive
+    fingerprint data, the verdict is DOWNGRADED to a model claim. (Omitting grounded_output
+    preserves the legacy grounded==verified behaviour for existing callers.)
     Negated / false-positive lines are skipped. Captures any non-negated line asserting a
     positive verdict even if no severity word shares that line.
     """
     NEG = ("NOT VULNERABLE", "NOT EXPLOITABLE", "NO CRITICAL", "NO HIGH", "NOT CONFIRMED",
            "UNABLE TO CONFIRM", "NOTHING CONFIRMED", "NO VULNERABILIT", "NOT CONFIRM",
            "FALSE POSITIVE")
     POS = ("CONFIRMED", "EXPLOITABLE", "VULNERABLE")
-    tag = "[VERIFIED — grounded run]" if grounded else "[MODEL CLAIM — verify PoC]"
+    effective_grounded = grounded
+    if grounded and grounded_output is not None and _grounded_output_is_passive_only(grounded_output):
+        effective_grounded = False
+    tag = "[VERIFIED — grounded run]" if effective_grounded else "[MODEL CLAIM — verify PoC]"
     out, seen = [], set()
     for line in response.split("\n"):
         s = line.strip().lstrip("#>*-• ").strip()
@@ -831,6 +879,9 @@ def run_brain_scanner(target: str, briefing: str = "", cookies: str = "",
     findings = []
     iteration = 0
     successful_runs = 0   # scripts that actually executed (no syntax/tooling error)
+    grounded_stdout = ""  # F9: accumulated stdout of grounded runs, so a CONFIRMED
+                          # verdict is only tagged [VERIFIED] when SOME grounded output
+                          # actually corroborates it (not just a passive server banner).
     empty_streak = 0      # consecutive empty provider responses (see MAX_EMPTY_STREAK)
 
     mode_labels = {
@@ -903,7 +954,8 @@ def run_brain_scanner(target: str, briefing: str = "", cookies: str = "",
                 # [VERIFIED ...] (reporter keeps it); negated/FP lines are skipped. Fixes the
                 # bug where a grounded confirmation vanished as 'Findings: 0' because the
                 # severity word and 'CONFIRMED' landed on separate lines.
-                findings.extend(_verdict_findings(response, grounded=successful_runs > 0))
+                findings.extend(_verdict_findings(response, grounded=successful_runs > 0,
+                                                  grounded_output=grounded_stdout))
                 break
             else:
                 # Ask brain to write code
@@ -982,6 +1034,10 @@ def run_brain_scanner(target: str, briefing: str = "", cookies: str = "",
                 # gate; the verdict logic below refuses to finish while successful_runs==0).
                 if _is_grounded_run(result):
                     successful_runs += 1
+                    # F9: keep the grounded stdout (capped) so the verdict tagger can
+                    # confirm the CONFIRMED claim rests on real evidence, not a banner.
+                    if len(grounded_stdout) < 20000:
+                        grounded_stdout += (result.get("stdout") or "") + "\n"
                 # Grounded findings: only from ACTUAL script stdout. Plain substring
                 # matching records negative lines ("NOT VULNERABLE", "No critical ...")
                 # as findings — skip any line carrying a negation marker. Markers are

har_vapt_engine.py

@@ -396,26 +396,70 @@ def test_xss(self) -> Dict:
                                           evidence=ctx, param=param, payload=payload)
                                 results['vulnerable'].append(url)
                                 break
-                        # SSTI
-                        if payload == '{{7*7}}' and '49' in body:
-                            self._log('critical', 'SSTI', url,
-                                      f"Template expression evaluated in param '{param}'",
-                                      param=param, payload=payload)
-                            results['vulnerable'].append(url)
-                            break
-                        if payload == '${7*7}' and '49' in body:
-                            self._log('critical', 'SSTI (EL)', url,
-                                      f"EL expression evaluated in param '{param}'",
-                                      param=param, payload=payload)
-                            results['vulnerable'].append(url)
-                            break
                     except Exception:
                         pass
+                # SSTI is confirmed by a dedicated evaluation probe (NOT the '49 in
+                # body' substring heuristic, which fired on any page containing 49 —
+                # friends full-tool review F4). Run once per param.
+                if self._probe_ssti(url, method, base_params, param):
+                    results['vulnerable'].append(url)
             results['tested'] += 1
-
         self.test_results['xss'] = results
         return results
 
+    def _probe_ssti(self, url: str, method: str, base_params: dict, param: str) -> bool:
+        """Confirm server-side template / EL injection by EVALUATION, not substring.
+
+        friends full-tool review F4: the old check flagged CRITICAL SSTI whenever
+        ``49`` appeared in the body after injecting ``{{7*7}}`` — so any page with
+        a price/id/"49 results" was a fabricated CRITICAL. This probe instead:
+          1. uses a DISTINCTIVE arithmetic canary whose product is improbable in
+             normal content (coincidental substring match is negligible);
+          2. confirms the product is ABSENT from a baseline (un-injected) response,
+             so dynamic content that already contains the number can't false-fire;
+          3. requires the raw expression NOT to be reflected verbatim — a template
+             that is echoed unevaluated is reflection, not SSTI.
+        Returns True and logs a CRITICAL finding on confirmation.
+        """
+        base = url.split('?')[0]
+        # Distinctive operands (computed, so no manual-arithmetic risk).
+        a, b = 91193, 90007
+        product = str(a * b)
+
+        def _fetch(value):
+            params = {**base_params, param: value}
+            try:
+                if method == 'POST':
+                    r = self.session.post(base, data=params, timeout=15)
+                else:
+                    r = self.session.get(base, params=params, timeout=15)
+                return r.text or ""
+            except Exception:
+                return None
+
+        baseline = _fetch("vapt_ssti_baseline")
+        if baseline is None or product in baseline:
+            # Request failed, or the number already appears un-injected — cannot
+            # attribute a later match to evaluation. Do NOT fire (anti-fabrication).
+            return False
+
+        for expr_tmpl, label in (("{{%d*%d}}", "SSTI"), ("${%d*%d}", "SSTI (EL)")):
+            expr = expr_tmpl % (a, b)
+            body = _fetch(expr)
+            if body is None:
+                continue
+            # Evaluated: product present, AND neither the wrapped expression nor the
+            # bare "a*b" echoed back (those would be reflection, not evaluation).
+            if product in body and expr not in body and f"{a}*{b}" not in body:
+                idx = body.find(product)
+                self._log('critical', label, url,
+                          f"Template expression evaluated in param '{param}' "
+                          f"({a}*{b} rendered as {product})",
+                          evidence=body[max(0, idx - 40):idx + len(product) + 40],
+                          param=param, payload=expr)
+                return True
+        return False
+
     # ── Command Injection ─────────────────────────────────────────────────
 
     def test_command_injection(self) -> Dict: