Keep both accounts signed in for the cross account checks

The cross account checks run last, after the engines, which on a deep scan
can take hours.  By then the sign in from the start of the scan has often
run out, and the site answers an authenticated request with a "your
session has expired" page.  That page is HTTP 200, so the old check read
it as the account's real page and could report access that was not there.

Three changes fix it.  Each account signs in fresh at the start of this
stage, so the sessions are seconds old, not hours.  A "session expired" or
"sign in" page is now recognised and never treated as the account's
content, so a check never fires against one.  And if a session still runs
out while the checks are going, that account signs in again once and the
request is retried, so a slow stage does not lose the session.

Author: hackrange

app/core/dast_checks/engine.py

@@ -192,6 +192,21 @@ def _normalize(body: str) -> str:
     return re.sub(r"\s+", " ", body).strip()
 
 
+# A page that answers HTTP 200 but is really a sign in wall or a session that
+# has run out.  A long scan can outlive the session it signed in with, and the
+# site then answers these instead of the real page; a status check alone is
+# fooled, so a compare against one would be meaningless.
+_AUTHLESS = re.compile(
+    r"session (has )?(expired|timed ?out)|your session (has )?(expired|timed)|"
+    r"(log|sign) ?in again|authenticate again|please (authenticate|log ?in|sign ?in)|"
+    r"you need to (log|sign) ?in|(log|sign) ?in to continue", re.I)
+
+
+def _looks_authless(resp) -> bool:
+    """True when a response is a sign in or session expired wall, not content."""
+    return bool(_AUTHLESS.search(((resp or {}).get("body") or "")[:8000]))
+
+
 def _env_for(response: dict, control: dict, oob_hit: bool, saved: dict = None) -> dict:
     """The names and helpers a differential may use.
 
@@ -524,6 +539,23 @@ def _run_two_account(spec: dict, ctx: dict, domains, budget: Budget) -> list:
     a = Client(base_url, domains, "read", budget, allow_private, ctx.get("headers"))
     b = Client(base_url, domains, "read", budget, allow_private, ctx.get("second_headers"))
     anon = Client(base_url, domains, "read", budget, allow_private, None)
+    # A long scan can outlive the sign in, so each account can sign in again once
+    # if its session has run out while the checks run.  Kept to one per account
+    # so a site that will not keep a session is not hammered.
+    reauth = {"a": ctx.get("reauth_a"), "b": ctx.get("reauth_b")}
+    may_reauth = {"a": bool(reauth["a"]), "b": bool(reauth["b"])}
+
+    def live(client, who, url):
+        """Fetch as one account; if the session has run out, sign in again once
+        and retry, so no check ever runs against a session expired page."""
+        resp = client.request("GET", url)
+        if _looks_authless(resp) and may_reauth[who]:
+            may_reauth[who] = False
+            fresh = reauth[who]()
+            if fresh:
+                client.headers = dict(fresh)
+                resp = client.request("GET", url)
+        return resp
 
     confirm = spec.get("confirm") or {}
     differential = confirm.get("differential") or "False"
@@ -533,18 +565,25 @@ def _run_two_account(spec: dict, ctx: dict, domains, budget: Budget) -> list:
     for url in _b_objects(b, base_url, ctx.get("code_routes")):
         confirmations, resp_a = 0, None
         for _attempt in range(retest):
-            resp_b = b.request("GET", url)
-            if resp_b.get("status") != 200:
+            resp_b = live(b, "b", url)
+            # It must be the second account's real object: a 200 that is not a
+            # sign in or session expired wall.
+            if resp_b.get("status") != 200 or _looks_authless(resp_b):
                 break
-            resp_a = a.request("GET", url)
+            resp_a = live(a, "a", url)
             resp_anon = anon.request("GET", url)
+            # The first account must have reached the real object too; a sign in
+            # wall is not access and there would be nothing to compare.
+            if _looks_authless(resp_a):
+                resp_a = None
+                break
             env = _env_for(resp_a, {}, False,
                            saved={"b": resp_b, "anon": resp_anon})
             if evaluate(differential, env):
                 confirmations += 1
             else:
                 break
-        if confirmations == retest and confirmations > 0:
+        if confirmations == retest and confirmations > 0 and resp_a is not None:
             f = _finding(spec, a, resp_a, url)
             f["evidence"] = _two_account_evidence(url, resp_a, resp_b, resp_anon)
             f["reproduce"] = _two_account_reproduce(url)

app/core/external.py

@@ -2189,7 +2189,7 @@ def _run_curated_rules(scan_id, asset_id, url, wb, session, opts, rulesets) -> t
 
     specs, domains = [], []
     excluded = invalid = 0
-    second_headers = None
+    auth_a = auth_b = None
     with session_scope() as db:
         for row in db.query(DastRule).filter(DastRule.enabled.is_(True)).all():
             if not profiles.selected(row.rule_class, row.rulesets, rulesets):
@@ -2203,16 +2203,28 @@ def _run_curated_rules(scan_id, asset_id, url, wb, session, opts, rulesets) -> t
         domains = scope.company_domains(
             settings_mod.get(db, "web.company_domains") or [])
         oob = callback.for_scan(db)
-        # The second account, for the compare checks.  Built only when one is
-        # set and at least one enabled rule needs it.
-        if any("two_accounts" in (s.get("needs") or []) for s in specs):
-            asset = db.get(ExternalAsset, asset_id)
-            auth_b = external_auth.load_b(asset) if asset else {}
-            if auth_b:
-                sess_b = external_auth.session_for(auth_b, opts["allow_private"])
-                if sess_b.get("ok"):
-                    second_headers = _session_headers(sess_b)
-    headers = _session_headers(session)
+        asset = db.get(ExternalAsset, asset_id)
+        auth_a = (external_auth.load(asset) or None) if asset else None
+        # The second account, for the compare checks.  Loaded only when at least
+        # one enabled rule needs it.
+        if asset and any("two_accounts" in (s.get("needs") or []) for s in specs):
+            auth_b = external_auth.load_b(asset) or None
+
+    # Sign in FRESH for this stage.  The curated rules run last, after the engines
+    # (which on premium can take hours), by when the sign in from the start of the
+    # scan has usually run out and the site answers "session expired".  Each
+    # account is signed in again now, and the engine can sign in again once more
+    # if a session still lapses while the checks run.
+    def _fresh_headers(auth):
+        if not auth:
+            return None
+        s = external_auth.session_for(auth, opts["allow_private"])
+        return _session_headers(s) if s.get("ok") else None
+
+    headers = _fresh_headers(auth_a) or _session_headers(session)
+    second_headers = _fresh_headers(auth_b)
+    reauth_a = (lambda: _fresh_headers(auth_a)) if auth_a else None
+    reauth_b = (lambda: _fresh_headers(auth_b)) if auth_b else None
 
     host = urlparse(url).hostname or ""
     row = {"tool": "curated", "name": "Premium curated rules", "status": "ok",
@@ -2231,6 +2243,7 @@ def _run_curated_rules(scan_id, asset_id, url, wb, session, opts, rulesets) -> t
            "allow_private": opts["allow_private"],
            "code_routes": list(wb.get("routes") or []),
            "headers": headers or None, "second_headers": second_headers,
+           "reauth_a": reauth_a, "reauth_b": reauth_b,
            "budget": 60, "oob": oob}
     from collections import Counter