Two account support, and the compare checks that use it

A site can now hold a second sign in, a second test account, alongside the
first.  The second account is stored encrypted in its own slot, set from the
site page the same way as the first, and never shown again once saved.

With two accounts set, the enhanced engine runs the compare checks safely and
read only: it finds an object that belongs to the second account from that
account's own pages, then reads it as the first account and as a signed out
visitor.  It reports only when the first account can read the second account's
object and a signed out visitor cannot, which is broken object access.  It
stays silent when the site scopes the object to its owner, when the object is
public, and when no second account is set.

Three curated rules now run on this: cross account read of a numeric id, of an
opaque id, and the general case.  They need no source and no ids supplied; the
object addresses come from the second account's own pages.  The rule that
needed an admin account is set aside until that is modelled.

The differential grammar gains the helpers these checks need: compare two
saved responses, and read a saved response's status.

Author: hackrange

app/core/dast_checks/curated/access-control-bola-uuid.json

@@ -3,27 +3,26 @@
   "class": "access-control",
   "safety": "read",
   "severity_floor": "high",
-  "title": "Opaque id can be substituted to read another tenant's object (needs two accounts)",
-  "summary": "A UUID or opaque id can be swapped for another to read an object that is not the caller's.  Needs a second account to prove, so it skips until two accounts are configured.",
+  "title": "A UUID or opaque object id can be read by another account",
+  "summary": "An object named by a UUID or opaque id that belongs to the second account can be read by the first account.  Guessing the id is hard, but access must still be checked.  Needs a second account configured for the site.",
   "needs": [
-    "two_accounts",
-    "code_routes"
+    "two_accounts"
   ],
   "probe": [
     {
       "request": {
         "method": "GET",
-        "path": "/{route}"
+        "path": "/"
       }
     }
   ],
   "confirm": {
     "control": "A path that cannot exist, so the site's not found answer is known.",
-    "differential": "status==200 and body_differs(control)",
+    "differential": "status==200 and status_of(b)==200 and same(response, b) and differ(response, anon)",
     "retest": 2
   },
-  "reproduce": "As account A, request an object id known to belong to account B.",
-  "remediate": "Check object ownership on the server for every object level read.",
+  "reproduce": "Sign in as account A and open an object URL (with account B's UUID) that belongs to account B.  If A sees B's object, object access is broken.",
+  "remediate": "Check ownership on the server for every object, not just that the id is hard to guess.",
   "references": [],
   "mappings": {
     "cwe": "CWE-639",

app/core/dast_checks/curated/access-control-horizontal-ab.json

@@ -3,27 +3,26 @@
   "class": "access-control",
   "safety": "read",
   "severity_floor": "high",
-  "title": "One account can read another account's data (needs two accounts)",
-  "summary": "Account A can read what only account B should see.  This is the cleanest proof of broken object authorization, and it needs both accounts.",
+  "title": "One account can read another account's object",
+  "summary": "The first account can read an object that belongs to the second account, which a signed out visitor cannot.  This is broken object level authorization.  Needs a second account configured for the site.",
   "needs": [
-    "two_accounts",
-    "code_routes"
+    "two_accounts"
   ],
   "probe": [
     {
       "request": {
         "method": "GET",
-        "path": "/{route}"
+        "path": "/"
       }
     }
   ],
   "confirm": {
     "control": "A path that cannot exist, so the site's not found answer is known.",
-    "differential": "status==200 and body_differs(control)",
+    "differential": "status==200 and status_of(b)==200 and same(response, b) and differ(response, anon)",
     "retest": 2
   },
-  "reproduce": "Fetch the same route as account A and account B and compare the bodies.",
-  "remediate": "Scope every query to the signed in account.",
+  "reproduce": "Sign in as account A.  Open an object URL that belongs to account B (for example /invoice/<B's id>).  If A sees B's data, object access is broken.",
+  "remediate": "Check that the signed in account owns the object on every object read.",
   "references": [],
   "mappings": {
     "cwe": "CWE-639",

app/core/dast_checks/curated/access-control-idor-numeric.json

@@ -3,27 +3,26 @@
   "class": "access-control",
   "safety": "read",
   "severity_floor": "high",
-  "title": "Numeric object id can be walked to another user's record (needs two accounts)",
-  "summary": "A numeric id in a route can be changed to read a record that belongs to someone else.  Proving this needs a second test account, so it only runs once two accounts are configured.",
+  "title": "A numeric object id can be read by another account",
+  "summary": "An object named by a numeric id that belongs to the second account can be read by the first account, so ids can be walked to read other records.  Needs a second account configured for the site.",
   "needs": [
-    "two_accounts",
-    "code_routes"
+    "two_accounts"
   ],
   "probe": [
     {
       "request": {
         "method": "GET",
-        "path": "/{route}"
+        "path": "/"
       }
     }
   ],
   "confirm": {
     "control": "A path that cannot exist, so the site's not found answer is known.",
-    "differential": "status==200 and body_differs(control)",
+    "differential": "status==200 and status_of(b)==200 and same(response, b) and differ(response, anon)",
     "retest": 2
   },
-  "reproduce": "As account A, request account B's object id over the same route.",
-  "remediate": "Check the object's owner against the signed in user on every read.",
+  "reproduce": "Sign in as account A and open an object URL with account B's numeric id (for example /account/<B's id>).  If A sees B's record, access is broken.",
+  "remediate": "Check the object's owner against the signed in account on every read.",
   "references": [],
   "mappings": {
     "cwe": "CWE-639",

app/core/dast_checks/curated/access-control-vertical-privilege.json

@@ -22,6 +22,7 @@
 from __future__ import annotations
 
 import ast
+import re
 import secrets
 import time
 from urllib.parse import urljoin, urlparse
@@ -104,7 +105,7 @@ def request(self, method: str, path: str, query: dict = None,
 # ------------------------------------------------------------------ the DSL
 
 _ALLOWED_FUNCS = {"body_differs", "body_contains", "header", "reflected",
-                  "status_in"}
+                  "status_in", "differ", "same", "status_of"}
 _ALLOWED_CMP = (ast.Eq, ast.NotEq, ast.Lt, ast.LtE, ast.Gt, ast.GtE,
                 ast.In, ast.NotIn)
 
@@ -187,8 +188,12 @@ def _normalize(body: str) -> str:
     return re.sub(r"\s+", " ", body).strip()
 
 
-def _env_for(response: dict, control: dict, oob_hit: bool) -> dict:
-    """The names and helpers a differential may use."""
+def _env_for(response: dict, control: dict, oob_hit: bool, saved: dict = None) -> dict:
+    """The names and helpers a differential may use.
+
+    saved holds other responses by name (for example b and anon in a two
+    account check), so a differential can compare them.
+    """
     body = response.get("body") or ""
     headers = response.get("headers") or {}
 
@@ -207,14 +212,27 @@ def reflected(marker):
     def status_in(*codes):
         return response.get("status") in codes
 
-    return {
+    def differ(a, b):
+        return _normalize((a or {}).get("body") or "") != _normalize((b or {}).get("body") or "")
+
+    def same(a, b):
+        return not differ(a, b)
+
+    def status_of(resp):
+        return (resp or {}).get("status", 0)
+
+    env = {
         "status": response.get("status", 0),
+        "response": response,
         "control": control,
         "oob_hit": oob_hit,
         "_funcs": {"body_differs": body_differs, "body_contains": body_contains,
                    "header": header, "reflected": reflected,
-                   "status_in": status_in},
+                   "status_in": status_in, "differ": differ, "same": same,
+                   "status_of": status_of},
     }
+    env.update(saved or {})
+    return env
 
 
 # ------------------------------------------------------------------ running
@@ -277,6 +295,18 @@ def run_rule(spec: dict, ctx: dict) -> dict:
         return out
 
     budget = Budget(int(ctx.get("budget") or DEFAULT_BUDGET))
+
+    # A rule that compares two accounts runs its own path: it reads an object
+    # that belongs to the second account and checks the first can read it too.
+    if "two_accounts" in (spec.get("needs") or []):
+        try:
+            out["findings"] = _run_two_account(spec, ctx, domains, budget)
+        except Refused as exc:
+            out["status"], out["detail"] = "refused", str(exc)
+        except Exception as exc:  # noqa: BLE001
+            out["status"], out["detail"] = "error", f"{type(exc).__name__}: {exc}"[:200]
+        return out
+
     client = Client(base_url, domains, spec["safety"], budget,
                     allow_private=bool(ctx.get("allow_private")),
                     headers=ctx.get("headers"))
@@ -320,6 +350,72 @@ def run_rule(spec: dict, ctx: dict) -> dict:
     return out
 
 
+_OBJECT_PATH = re.compile(r"/\d{1,12}(?:/|$)|/[0-9a-fA-F]{8,}(?:/|$)"
+                          r"|[?&][\w-]*id=\w+", re.I)
+
+
+def _b_objects(client_b: Client, base_url: str, code_routes) -> list:
+    """Addresses that look like they point at a specific object of account b.
+
+    A cross account read is only meaningful against an object that belongs to
+    the other account, so this keeps the links from account b's own pages (and
+    any code routes) that carry an id, and leaves the generic pages out.
+    """
+    out = []
+    try:
+        landing = client_b.request("GET", "/")
+    except Refused:
+        return out
+    from app.core.dast_engines import links_in
+
+    for url in links_in(base_url, landing.get("body") or ""):
+        if _OBJECT_PATH.search(url) and url not in out:
+            out.append(url)
+    for route in code_routes or []:
+        if _OBJECT_PATH.search(route) and route not in out:
+            out.append(route)
+    return out[:MAX_ROUTES]
+
+
+def _run_two_account(spec: dict, ctx: dict, domains, budget: Budget) -> list:
+    """Account A reads an object that belongs to account B.  -> findings.
+
+    Reads only.  For each object address found in B's own pages: fetch it as B
+    (to confirm it is really B's and it is a real page), as A, and signed out.
+    The differential decides, for example: A got 200, it is really B's object,
+    A sees the same body as B, and the signed out visitor is turned away.
+    """
+    base_url = ctx.get("base_url") or ""
+    allow_private = bool(ctx.get("allow_private"))
+    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)
+
+    confirm = spec.get("confirm") or {}
+    differential = confirm.get("differential") or "False"
+    retest = int(confirm.get("retest") or 1)
+
+    findings = []
+    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:
+                break
+            resp_a = a.request("GET", url)
+            resp_anon = anon.request("GET", url)
+            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:
+            findings.append(_finding(spec, a, resp_a, url))
+            break   # one solid finding per rule
+    return findings
+
+
 def _run_probe(spec: dict, client: Client, route: str, oob_value: str) -> dict:
     """Send the probe steps; the last response is what the differential reads.
 

app/core/dast_checks/engine.py

@@ -484,6 +484,10 @@ def bulk_update(db, assets: list, changes: dict, actor: str, user=None) -> list:
             if changes.get("auth") is not None:
                 kind, values = changes["auth"]
                 external_auth.save(asset, kind, values, actor, asset.parent_domain)
+            if changes.get("auth_b") is not None:
+                kind, values = changes["auth_b"]
+                external_auth.save(asset, kind, values, actor, asset.parent_domain,
+                                   slot="b")
             if repos:
                 mode, picks = repos
                 if mode == "add":
@@ -1898,7 +1902,19 @@ def _scan_rulesets(scan_id, profile) -> list:
     return ["*"] if profile == "premium" else []
 
 
-def _run_curated_rules(scan_id, url, wb, session, opts, rulesets) -> tuple:
+def _session_headers(sess: dict) -> dict:
+    """The request headers a built session signs in with: cookie and any headers."""
+    out = {}
+    if not sess:
+        return out
+    if sess.get("cookie"):
+        out["Cookie"] = sess["cookie"]
+    for name, value in sess.get("headers") or []:
+        out[name] = value
+    return out
+
+
+def _run_curated_rules(scan_id, asset_id, url, wb, session, opts, rulesets) -> tuple:
     """Run the enabled curated rules in the chosen rulesets against the site.
 
     rulesets is the list the scan profile chose (["*"] for all).  -> (findings,
@@ -1908,12 +1924,14 @@ def _run_curated_rules(scan_id, url, wb, session, opts, rulesets) -> tuple:
     """
     from urllib.parse import urlparse
 
+    from app.core import external_auth
     from app.core import settings_store as settings_mod
     from app.core.dast_checks import callback, engine, profiles, scope
     from app.core.dast_checks import spec as spec_mod
-    from app.core.models import DastRule
+    from app.core.models import DastRule, ExternalAsset
 
-    specs, domains, headers = [], [], {}
+    specs, domains = [], []
+    second_headers = 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):
@@ -1925,8 +1943,16 @@ def _run_curated_rules(scan_id, url, wb, session, opts, rulesets) -> tuple:
         domains = scope.company_domains(
             settings_mod.get(db, "web.company_domains") or [])
         oob = callback.for_scan(db)
-    if session.get("cookie"):
-        headers["Cookie"] = session["cookie"]
+        # 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)
 
     host = urlparse(url).hostname or ""
     row = {"tool": "curated", "name": "Premium curated rules", "status": "ok",
@@ -1944,7 +1970,8 @@ def _run_curated_rules(scan_id, url, wb, session, opts, rulesets) -> tuple:
     ctx = {"base_url": url, "host": host, "domains": domains,
            "allow_private": opts["allow_private"],
            "code_routes": list(wb.get("routes") or []),
-           "headers": headers or None, "budget": 60, "oob": oob}
+           "headers": headers or None, "second_headers": second_headers,
+           "budget": 60, "oob": oob}
     started = time.monotonic()
     findings, ran, skipped, confirmed = [], 0, 0, 0
     for spec in specs:
@@ -2060,8 +2087,8 @@ def _builtin_run(scan_id, host, asset_id, profile, opts, auth, proxy) -> None:
     if url and plan:
         rulesets = _scan_rulesets(scan_id, profile)
         if rulesets:
-            rule_findings, rule_row = _run_curated_rules(scan_id, url, wb, session,
-                                                         opts, rulesets)
+            rule_findings, rule_row = _run_curated_rules(scan_id, asset_id, url, wb,
+                                                         session, opts, rulesets)
             findings += rule_findings
             engines.append(rule_row)
             _update(scan_id, engines=json.dumps(engines))