Release 1.3.0

README.md

@@ -44,6 +44,7 @@ Passive modules analyze cryptographic products (cookies, tokens, signed URLs, et
 | Yii2_SignedCookies | Checks Yii2 framework signed cookies for known cookie validation keys |
 | Shiro_RememberMe | Checks Apache Shiro `rememberMe` cookies for known AES encryption keys |
 | LTPA_Token | Checks IBM WebSphere `LtpaToken` and `LtpaToken2` cookies for known LTPA encryption keys |
+| NextAuth | Checks NextAuth.js / Auth.js JWE session cookies for a known or weak `NEXTAUTH_SECRET` / `AUTH_SECRET` |
 
 ### Active Modules
 
@@ -350,6 +351,7 @@ Rack2_SignedCookies = modules_loaded["rack2_signedcookies"]
 Yii2_SignedCookies = modules_loaded["yii2_signedcookies"]
 Shiro_RememberMe = modules_loaded["shiro_rememberme"]
 LTPA_Token = modules_loaded["ltpa_token"]
+NextAuth = modules_loaded["nextauth"]
 
 
 x = ASPNET_Viewstate()
@@ -519,6 +521,17 @@ if r:
 else:
     print("KEY NOT FOUND :(")
 
+x = NextAuth()
+print(f"###{str(x.__class__.__name__)}###")
+r = x.check_secret(
+    "eyJhbGciOiJkaXIiLCJlbmMiOiJBMjU2R0NNIn0..eV7_ge7JJ9vqaYxW.97XOYXr0ANwWherKQ3wIwyLNBN7-A8O40pNSwihk4BPIDWUn3KoXzX5I9fV9rhmlkaILza1p3jVKhzcG"
+    "ISkE3nmx_gaxnXv6UlNfg2vMeA8A_jeQb9x9MgK1yBuIG_V-cw.5b2T1NpI1p4rku2w9mXZ-Q"
+)
+if r:
+    print(r)
+else:
+    print("KEY NOT FOUND :(")
+
 ```
 
 #### Carve

badsecrets/__version__.py

@@ -1 +1 @@
-__version__ = "1.2.1"
+__version__ = "1.3.0"

badsecrets/base.py

@@ -11,6 +11,14 @@
 
 log = logging.getLogger(__name__)
 
+# Deduplicated wordlists, cached per (custom_resource, resource_list) for the life of the process.
+# Wordlists are static during a run but check_secret() runs per event, so this turns thousands of
+# reads + dedup-set builds of a 250k-line list into one. Naturally bounded by the small set of
+# resource combinations. Under BBOT this lives in each persistent process-pool worker, so it
+# persists across events. A plain dict on purpose: auditable, no lru_cache.
+_resource_cache = {}
+
+
 generic_base64_regex = re.compile(
     r"^(?:[A-Za-z0-9+\/]{4}){8,}(?:[A-Za-z0-9+\/]{4}|[A-Za-z0-9+\/]{3}=|[A-Za-z0-9+\/]{2}={2})$"
 )
@@ -77,16 +85,31 @@ def _safe_hashcat(self, product):
             return None
 
     def load_resources(self, resource_list):
-        filepaths = []
-        if self.custom_resource:
-            filepaths.append(self.custom_resource)
-        for r in resource_list:
-            filepaths.append(f"{os.path.dirname(os.path.abspath(__file__))}/resources/{r}")
-        for filepath in filepaths:
-            with open(filepath) as r:
-                for line in r.readlines():
-                    if len(line) > 0:
-                        yield line
+        """Return the deduplicated lines of the given wordlists (plus any custom_resource).
+
+        Cached per (custom_resource, resource_list) so a 250k-line list is read and deduplicated
+        once per process instead of on every check_secret call. First-seen order is preserved.
+        """
+        key = (self.custom_resource, tuple(resource_list))
+        cached = _resource_cache.get(key)
+        if cached is None:
+            resource_dir = os.path.dirname(os.path.abspath(__file__))
+            filepaths = []
+            if self.custom_resource:
+                filepaths.append(self.custom_resource)
+            for r in resource_list:
+                filepaths.append(f"{resource_dir}/resources/{r}")
+            seen = set()
+            deduped = []
+            for filepath in filepaths:
+                with open(filepath) as f:
+                    for line in f:
+                        if len(line) > 0 and line not in seen:
+                            seen.add(line)
+                            deduped.append(line)
+            cached = tuple(deduped)
+            _resource_cache[key] = cached
+        return cached
 
     def carve_to_check_secret(self, s, **kwargs):
         if s.groups():
@@ -190,21 +213,25 @@ def _carve_body(self, body, cookies, headers, **kwargs):
         """Extract secrets from HTML body text. Override in subclasses for custom body carving."""
         results = []
         if self.carve_regex():
-            s = re.search(self.carve_regex(), body)
-            if s:
-                if not self.validate_carve or self.identify(s.groups()[0]):
-                    r = self.carve_to_check_secret(
-                        s, url=kwargs.get("url"), body=body, cookies=cookies, headers=headers
-                    )
-                    if r:
-                        r["type"] = "SecretFound"
-                    else:
-                        r = {"type": "IdentifyOnly"}
-                        r["hashcat"] = self._safe_hashcat(s.groups()[0])
-                    if "product" not in r:
-                        r["product"] = self.get_product_from_carve(s)
-                    r["location"] = "body"
-                    results.append(r)
+            # Walk every match rather than only the first. A page can carry several fields the
+            # carve regex matches -- an empty or decoy __VIEWSTATE alongside a real payload --
+            # and stopping at match #1 lets whichever appears first in the markup hide the rest.
+            for s in re.finditer(self.carve_regex(), body):
+                if self.validate_carve and not self.identify(s.groups()[0]):
+                    continue
+                r = self.carve_to_check_secret(s, url=kwargs.get("url"), body=body, cookies=cookies, headers=headers)
+                if r:
+                    r["type"] = "SecretFound"
+                else:
+                    r = {"type": "IdentifyOnly"}
+                    r["hashcat"] = self._safe_hashcat(s.groups()[0])
+                if "product" not in r:
+                    r["product"] = self.get_product_from_carve(s)
+                r["location"] = "body"
+                results.append(r)
+                # First candidate that identifies wins, keeping the one-result-per-body
+                # contract and avoiding repeat check_secret() work on expensive modules.
+                break
         return results
 
     @classmethod

badsecrets/helpers.py

@@ -4,9 +4,11 @@
 import hmac
 import struct
 import hashlib
+import base64
 import argparse
 import binascii
 from enum import Enum
+from Crypto.Cipher import AES
 from urllib.parse import urlparse
 from colorama import Fore, Style, init
 from badsecrets.errors import BadsecretsException
@@ -818,3 +820,70 @@ def get_apppaths_hashcodes(self):
         else:
             _, apppaths = self._extract_path_and_apppaths(self.url)
         return [dotnet_string_hashcode(apppath, self.db) for apppath in apppaths]
+
+
+def b64url_decode(data):
+    """URL-safe base64 decode that tolerates the stripped padding JOSE uses."""
+    if isinstance(data, str):
+        data = data.encode("ascii")
+    pad = -len(data) % 4
+    return base64.urlsafe_b64decode(data + (b"=" * pad))
+
+
+def hkdf_sha256(ikm, salt, info, length):
+    """RFC 5869 HKDF-Extract-and-Expand with SHA-256.
+
+    Kept product-agnostic: callers supply salt/info, because the derivation is the part each
+    product implements differently and cannot be recovered from a `dir` JWE.
+    """
+    if isinstance(ikm, str):
+        ikm = ikm.encode()
+    if isinstance(salt, str):
+        salt = salt.encode()
+    if isinstance(info, str):
+        info = info.encode()
+    prk = hmac.new(salt, ikm, hashlib.sha256).digest()
+    okm = b""
+    block = b""
+    counter = 1
+    while len(okm) < length:
+        block = hmac.new(prk, block + info + bytes([counter]), hashlib.sha256).digest()
+        okm += block
+        counter += 1
+    return okm[:length]
+
+
+def parse_jwe_compact(token):
+    """Split a compact-serialization JWE into its 5 segments, or return None if it isn't one."""
+    parts = token.split(".")
+    if len(parts) != 5:
+        return None
+    return tuple(parts)
+
+
+def jwe_decrypt(protected_b64, enc, cek, iv, ciphertext, tag):
+    """Decrypt one JWE given an already-derived content-encryption key.
+
+    Returns the plaintext bytes, or None if authentication fails. Supports the two `enc` values
+    NextAuth/Auth.js use (A256GCM and A256CBC-HS512). The AAD is the ASCII of the base64url
+    protected header, per RFC 7516.
+    """
+    aad = protected_b64.encode("ascii")
+    try:
+        if enc == "A256GCM":
+            cipher = AES.new(cek, AES.MODE_GCM, nonce=iv)
+            cipher.update(aad)
+            return cipher.decrypt_and_verify(ciphertext, tag)
+        if enc == "A256CBC-HS512":
+            if len(cek) != 64:
+                return None
+            mac_key, enc_key = cek[:32], cek[32:]
+            al = struct.pack(">Q", len(aad) * 8)
+            expected = hmac.new(mac_key, aad + iv + ciphertext + al, hashlib.sha512).digest()[:32]
+            if not hmac.compare_digest(expected, tag):
+                return None
+            cipher = AES.new(enc_key, AES.MODE_CBC, iv)
+            return unpad(cipher.decrypt(ciphertext))
+    except ValueError:
+        return None
+    return None

badsecrets/modules/passive/aspnet_compressedviewstate.py

@@ -12,13 +12,17 @@ class ASPNET_compressedviewstate(BadsecretsBase):
     yara_carve_rule = (
         "rule ASPNET_compressedviewstate_carve {"
         ' strings: $vs = "__VIEWSTATE" $vstate = "__VSTATE" $cvs = "__COMPRESSEDVIEWSTATE"'
-        " condition: $vs or $vstate or $cvs }"
+        ' $cvs_u = "__COMPRESSED_VSTATE"'
+        " condition: $vs or $vstate or $cvs or $cvs_u }"
     )
     description = {"product": "ASP.NET Compressed Viewstate", "secret": "unprotected", "severity": "CRITICAL"}
     carve_locations = ("body",)
 
     def carve_regex(self):
-        return re.compile(r"<input[^>]+__(?:VIEWSTATE|VSTATE|COMPRESSEDVIEWSTATE)\"\s*value=\"(.*?)\"")
+        return re.compile(
+            r"<input(?=[^>]*__(?:COMPRESSEDVIEWSTATE|COMPRESSED_VSTATE|VIEWSTATE|VSTATE)\")"
+            r"[^>]*?\svalue=\"(.*?)\""
+        )
 
     def check_secret(self, compressed_viewstate):
         if not self.identify(compressed_viewstate):

badsecrets/modules/passive/django_signedcookies.py

@@ -11,7 +11,7 @@ class DjangoSignedCookies(BadsecretsBase):
     def check_secret(self, django_signed_cookie):
         if not self.identify(django_signed_cookie):
             return False
-        for l in set(self.load_resources(["django_secret_keys.txt", "top_100000_passwords.txt"])):
+        for l in self.load_resources(["django_secret_keys.txt", "top_250000_passwords.txt"]):
             secret_key = l.rstrip()
             try:
                 r = djangoLoads(

badsecrets/modules/passive/express_signedcookies_cs.py

@@ -66,7 +66,7 @@ def check_secret(self, express_signed_cookie_data, *args):
         if not sig:
             return False
 
-        for l in set(self.load_resources(["express_session_secrets.txt", "top_100000_passwords.txt"])):
+        for l in self.load_resources(["express_session_secrets.txt", "top_250000_passwords.txt"]):
             secret = l.rstrip()
             r = self.expressVerify_cs(express_signed_cookie_data, sig, secret)
             if r:

badsecrets/modules/passive/express_signedcookies_es.py

@@ -59,7 +59,7 @@ def check_secret(self, express_signed_cookie):
         if not self.identify(express_signed_cookie):
             return False
 
-        for l in set(self.load_resources(["express_session_secrets.txt", "top_100000_passwords.txt"])):
+        for l in self.load_resources(["express_session_secrets.txt", "top_250000_passwords.txt"]):
             session_secret = l.rstrip()
 
             r = self.expressVerify_es(express_signed_cookie, session_secret)

badsecrets/modules/passive/flask_signedcookies.py

@@ -13,7 +13,7 @@ class Flask_SignedCookies(BadsecretsBase):
     def check_secret(self, flask_cookie):
         if not self.identify(flask_cookie):
             return None
-        for l in set(self.load_resources(["flask_secret_keys.txt", "top_100000_passwords.txt"])):
+        for l in self.load_resources(["flask_secret_keys.txt", "top_250000_passwords.txt"]):
             password = l.rstrip()
             r = flaskVerify(value=flask_cookie, secret=password)
             if r:

badsecrets/modules/passive/generic_jwt.py

@@ -103,7 +103,7 @@ def check_secret(self, JWT):
             return None
 
         if algorithm[0].lower() == "h":
-            for l in self.load_resources(["jwt_secrets.txt", "top_100000_passwords.txt"]):
+            for l in self.load_resources(["jwt_secrets.txt", "top_250000_passwords.txt"]):
                 key = l.strip()
 
                 r = self.jwtVerify(JWT, key, algorithm)

badsecrets/modules/passive/jsf_viewstate.py

@@ -28,7 +28,7 @@ class Jsf_viewstate(BadsecretsBase):
     carve_locations = ("body",)
 
     def carve_regex(self):
-        return re.compile(r"<input.+?name=\"javax\.faces\.ViewState\".+?value=\"([^\"]*)\"")
+        return re.compile(r"<input(?=[^>]*name=\"javax\.faces\.ViewState\")[^>]*?\svalue=\"([^\"]*)\"")
 
     # Mojarra 1.2.x - 2.0.3
     def DES3_decrypt(self, ct, password):
@@ -232,7 +232,7 @@ def check_secret(self, jsf_viewstate_value):
             else:
                 jsf_viewstate_value = base64.b64encode(uncompressed)
 
-        for l in set(self.load_resources(["jsf_viewstate_passwords.txt", "top_100000_passwords.txt"])):
+        for l in self.load_resources(["jsf_viewstate_passwords.txt", "top_250000_passwords.txt"]):
             with suppress(ValueError):
                 password = l.rstrip()
                 if self.DES3_decrypt(jsf_viewstate_value, password):

badsecrets/modules/passive/nextauth.py

@@ -0,0 +1,98 @@
+import re
+import json
+from badsecrets.base import BadsecretsBase
+from badsecrets.helpers import b64url_decode, parse_jwe_compact, hkdf_sha256, jwe_decrypt
+
+# NextAuth (v4) derives with an empty salt and this info string; v4 is the only variant that emits
+# A256GCM.
+V4_INFO = "NextAuth.js Generated Encryption Key"
+
+# Auth.js (v5) uses the session-cookie name as the HKDF salt and interpolates it into the info
+# string. check_secret() only receives the cookie value, not its name, so we try each default
+# name; custom cookie names are not covered. v5 always emits A256CBC-HS512.
+V5_COOKIE_NAMES = ("authjs.session-token", "__Secure-authjs.session-token")
+
+# (salt, info, key_length) derivations, precomputed once since the info strings never change.
+_GCM_DERIVATIONS = ((b"", V4_INFO, 32),)
+_CBC_DERIVATIONS = tuple((name.encode(), f"Auth.js Generated Encryption Key ({name})", 64) for name in V5_COOKIE_NAMES)
+# Auth.js v5 hardcodes enc=A256CBC-HS512 on encode, so a cookie-name-salt + A256GCM token cannot be
+# produced without patching Auth.js. A256GCM is therefore always v4 (empty salt) — there is no real
+# v5-GCM combination to try.
+_DERIVATIONS_BY_ENC = {"A256GCM": _GCM_DERIVATIONS, "A256CBC-HS512": _CBC_DERIVATIONS}
+
+
+class NextAuth(BadsecretsBase):
+    # A `dir` JWE has an empty encrypted-key segment, so the token is `header..iv.ct.tag`. The
+    # double dot is the tell that separates it from a 3-segment signed JWT.
+    identify_regex = re.compile(r"^eyJ[\w-]+\.\.[\w-]+\.[\w-]+\.[\w-]+$")
+    description = {
+        "product": "NextAuth.js / Auth.js Session Token",
+        "secret": "NEXTAUTH_SECRET / AUTH_SECRET",
+        "severity": "HIGH",
+    }
+    carve_locations = ("cookies",)
+
+    def carve(self, body=None, cookies=None, headers=None, http_response=None, **kwargs):
+        # NextAuth splits large session tokens across `<name>.0`, `<name>.1`, ... Reassemble those
+        # chunks before the normal carve so the full JWE is seen as one value.
+        if cookies:
+            cookies = self._reassemble_chunks(cookies)
+        return super().carve(body=body, cookies=cookies, headers=headers, http_response=http_response, **kwargs)
+
+    @staticmethod
+    def _reassemble_chunks(cookies):
+        chunk_regex = re.compile(r"^(?P<base>.*session-token)\.(?P<idx>\d+)$")
+        chunks = {}
+        result = {}
+        for name, value in cookies.items():
+            match = chunk_regex.match(name)
+            if match:
+                chunks.setdefault(match.group("base"), {})[int(match.group("idx"))] = value
+            else:
+                result[name] = value
+        for base, parts in chunks.items():
+            result[base] = "".join(parts[i] for i in sorted(parts))
+        return result
+
+    def _parse_token(self, token):
+        # Everything here is secret-independent, so check_secret() does it once rather than per
+        # candidate. Returns (protected_b64, enc, iv, ciphertext, tag, derivations) or None.
+        parsed = parse_jwe_compact(token)
+        if not parsed:
+            return None
+        protected_b64, _, iv_b64, ct_b64, tag_b64 = parsed
+        try:
+            header = json.loads(b64url_decode(protected_b64))
+            iv = b64url_decode(iv_b64)
+            ciphertext = b64url_decode(ct_b64)
+            tag = b64url_decode(tag_b64)
+        except (ValueError, json.JSONDecodeError):
+            return None
+        if header.get("alg") != "dir":
+            return None
+        derivations = _DERIVATIONS_BY_ENC.get(header.get("enc"))
+        if not derivations:
+            return None
+        return protected_b64, header["enc"], iv, ciphertext, tag, derivations
+
+    def check_secret(self, token):
+        if not self.identify(token):
+            return None
+        parsed = self._parse_token(token)
+        if not parsed:
+            return None
+        protected_b64, enc, iv, ciphertext, tag, derivations = parsed
+        for l in self.load_resources(["nextauth_secrets.txt", "top_250000_passwords.txt"]):
+            secret = l.strip()
+            if not secret:
+                continue
+            for salt, info, keylen in derivations:
+                cek = hkdf_sha256(secret, salt, info, keylen)
+                plaintext = jwe_decrypt(protected_b64, enc, cek, iv, ciphertext, tag)
+                if plaintext is not None:
+                    try:
+                        session = json.loads(plaintext)
+                    except (ValueError, json.JSONDecodeError):
+                        session = plaintext.decode(errors="replace")
+                    return {"secret": secret, "details": {"session": session, "enc": enc}}
+        return None