diff --git a/auth_utils.py b/auth_utils.py index de7f779..d55aefb 100644 --- a/auth_utils.py +++ b/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: diff --git a/autopilot_api_hunt.py b/autopilot_api_hunt.py index 4146793..38160a5 100644 --- a/autopilot_api_hunt.py +++ b/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 diff --git a/brain_scanner.py b/brain_scanner.py index fe13868..89de59f 100644 --- a/brain_scanner.py +++ b/brain_scanner.py @@ -441,14 +441,59 @@ 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=|