A python script to check if you are vulnerable to CVE-2025-1094

CVE-2025-1094-checker.py

#!/usr/bin/env python3
"""
PostgreSQL CVE-2025-1094 Vulnerability Checker

This script checks if a PostgreSQL installation is vulnerable to CVE-2025-1094,
a SQL injection vulnerability in PostgreSQL's escaping functions.

Usage:
  python3 check_postgres_cve_2025_1094.py [options]

Options:
  --host HOST           PostgreSQL host (default: localhost)
  --port PORT           PostgreSQL port (default: 5432)
  --user USER           PostgreSQL user (default: postgres)
  --password PASSWORD   PostgreSQL password (default: prompt)
  --dbname DBNAME       PostgreSQL database name (default: postgres)
  --verbose             Enable verbose output

Requirements:
  - Python 3.6+
  - psycopg2 module (pip install psycopg2-binary)
"""

import argparse
import getpass
import re
import sys
import time
from typing import Tuple

try:
    import psycopg2
    from psycopg2.extensions import connection
except ImportError:
    print("Error: psycopg2 module not found. Install it with 'pip install psycopg2-binary'")
    sys.exit(1)

# ANSI color codes
class Colors:
    HEADER = '\033[95m'
    BLUE = '\033[94m'
    CYAN = '\033[96m'
    GREEN = '\033[92m'
    YELLOW = '\033[93m'
    RED = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'
    UNDERLINE = '\033[4m'

# Define vulnerable versions
VULNERABLE_VERSIONS = {
    "13": lambda minor: minor < 19,
    "14": lambda minor: minor < 16,
    "15": lambda minor: minor < 11,
    "16": lambda minor: minor < 7,
    "17": lambda minor: minor < 3
}

def parse_version(version_str: str) -> Tuple[int, int]:
    """Parse PostgreSQL version string into major and minor versions."""
    match = re.search(r'PostgreSQL (\d+)\.(\d+)', version_str)
    if match:
        return int(match.group(1)), int(match.group(2))
    return 0, 0

def is_version_vulnerable(major: int, minor: int) -> bool:
    """Check if PostgreSQL version is vulnerable based on version numbers."""
    if str(major) in VULNERABLE_VERSIONS:
        return VULNERABLE_VERSIONS[str(major)](minor)
    return False

def get_safe_minor_version(major: int) -> int:
    """Get the safe minor version for a given major version."""
    version_map = {
        13: 19,
        14: 16,
        15: 11,
        16: 7,
        17: 3
    }
    return version_map.get(major, 0)

def check_server_version(conn: connection, verbose: bool = False) -> bool:
    """Check if the PostgreSQL server version is vulnerable."""
    try:
        with conn.cursor() as cur:
            cur.execute("SHOW server_version")
            full_version = cur.fetchone()[0]
            
            # Extract version numbers from string like "15.4" or "15.4 (Ubuntu 15.4-2.pgdg22.04+1)"
            match = re.search(r'^(\d+)\.(\d+)', full_version)
            if match:
                major = int(match.group(1))
                minor = int(match.group(2))
                
                if verbose:
                    print(f"{Colors.CYAN}PostgreSQL server version: {full_version}{Colors.ENDC}")
                    print(f"{Colors.CYAN}Parsed version: {major}.{minor}{Colors.ENDC}")
                
                if is_version_vulnerable(major, minor):
                    print(f"{Colors.RED}[VULNERABLE] PostgreSQL {major}.{minor} is affected by CVE-2025-1094{Colors.ENDC}")
                    print(f"{Colors.RED}             Update to at least version {major}.{get_safe_minor_version(major)}{Colors.ENDC}")
                    return True
                else:
                    print(f"{Colors.GREEN}[OK] PostgreSQL {major}.{minor} is not vulnerable to CVE-2025-1094{Colors.ENDC}")
                    return False
            else:
                print(f"{Colors.YELLOW}[WARNING] Could not parse PostgreSQL version: {full_version}{Colors.ENDC}")
                return True
    except psycopg2.Error as e:
        print(f"{Colors.RED}Error checking server version: {e}{Colors.ENDC}")
        return True

def print_ascii_art():
    """Print PostgreSQL elephant ASCII art."""
    elephant = f"""
{Colors.CYAN}    .--.
   |o_o |    {Colors.BOLD}PostgreSQL{Colors.ENDC}{Colors.CYAN}
   |:_/ |    {Colors.BOLD}Security Scanner{Colors.ENDC}{Colors.CYAN}
  //   \\ \\
 (|     | )
/'\_   _/`\\
\___)=(___/{Colors.ENDC}
"""
    print(elephant)

def main() -> None:
    parser = argparse.ArgumentParser(description="Check if PostgreSQL is vulnerable to CVE-2025-1094")
    parser.add_argument("--host", default="localhost", help="PostgreSQL host (default: localhost)")
    parser.add_argument("--port", type=int, default=5432, help="PostgreSQL port (default: 5432)")
    parser.add_argument("--user", default="postgres", help="PostgreSQL user (default: postgres)")
    parser.add_argument("--password", help="PostgreSQL password (default: prompt)")
    parser.add_argument("--dbname", default="postgres", help="PostgreSQL database (default: postgres)")
    parser.add_argument("--verbose", action="store_true", help="Enable verbose output")
    
    args = parser.parse_args()
    
    # Get password if not provided
    password = args.password
    if password is None:
        password = getpass.getpass(f"Password for user {args.user}: ")
    
    print_ascii_art()
    
    print(f"{Colors.BOLD}{Colors.HEADER}" + "=" * 70 + f"{Colors.ENDC}")
    print(f"{Colors.BOLD}{Colors.HEADER} PostgreSQL CVE-2025-1094 Vulnerability Checker{Colors.ENDC}")
    print(f"{Colors.BOLD}{Colors.HEADER}" + "=" * 70 + f"{Colors.ENDC}")
    print(f"{Colors.BLUE} This tool checks PostgreSQL for escaping function vulnerabilities (CVE-2025-1094)")
    print(f" It performs only safe, non-exploitative tests to determine vulnerability status{Colors.ENDC}")
    print(f"{Colors.BOLD}{Colors.HEADER}" + "=" * 70 + f"{Colors.ENDC}")
    print(f"{Colors.BOLD}Checking PostgreSQL at {args.host}:{args.port}{Colors.ENDC}")
    print(f"{Colors.BLUE}" + "-" * 70 + f"{Colors.ENDC}")
    
    # Connect to the database and check server
    try:
        print(f"{Colors.YELLOW}Connecting to PostgreSQL server...{Colors.ENDC}")
        conn = psycopg2.connect(
            host=args.host,
            port=args.port,
            user=args.user,
            password=password,
            dbname=args.dbname
        )
        conn.autocommit = True
        print(f"{Colors.GREEN}Connection established successfully{Colors.ENDC}")
        
        # Check server version
        print(f"\n{Colors.BLUE}Checking PostgreSQL server version:{Colors.ENDC}")
        is_vulnerable_by_version = check_server_version(conn, args.verbose)
        
        conn.close()
        
        # Final vulnerability assessment
        print(f"\n{Colors.BOLD}{Colors.HEADER}" + "=" * 70 + f"{Colors.ENDC}")
        print(f"{Colors.BOLD}{Colors.HEADER}VULNERABILITY ASSESSMENT:{Colors.ENDC}")
        print(f"{Colors.BOLD}{Colors.HEADER}" + "=" * 70 + f"{Colors.ENDC}")
        
        # Determine vulnerability status
        if is_vulnerable_by_version:
            vulnerability_status = "VULNERABLE"
            print(f"{Colors.BOLD}{Colors.RED}[!] RESULT: {vulnerability_status} - This PostgreSQL version is affected by CVE-2025-1094{Colors.ENDC}")
            
            alert_box = f"""
{Colors.RED}╔{'═' * 68}╗
║{' ' * 68}║
║  {Colors.BOLD}⚠️  WARNING: Security Vulnerability Detected  ⚠️{Colors.ENDC}{Colors.RED}{' ' * 12}║
║{' ' * 68}║
╚{'═' * 68}╝{Colors.ENDC}
"""
            print(alert_box)
            
            print(f"{Colors.YELLOW}\nDetails:{Colors.ENDC}")
            print(f"{Colors.YELLOW}- The PostgreSQL version you are running is in the affected version range")
            print(f"- All PostgreSQL versions prior to 13.19, 14.16, 15.11, 16.7, and 17.3 are vulnerable{Colors.ENDC}")
            
            print(f"{Colors.GREEN}\nRecommendation:{Colors.ENDC}")
            print(f"{Colors.GREEN}- Update to the latest patched version:")
            print(f"  * PostgreSQL 13.x: Update to 13.19 or later")
            print(f"  * PostgreSQL 14.x: Update to 14.16 or later")
            print(f"  * PostgreSQL 15.x: Update to 15.11 or later")
            print(f"  * PostgreSQL 16.x: Update to 16.7 or later")
            print(f"  * PostgreSQL 17.x: Update to 17.3 or later{Colors.ENDC}")
            
            print(f"{Colors.CYAN}\nAdditional Security Measures:{Colors.ENDC}")
            print(f"{Colors.CYAN}1. Validate and sanitize all inputs before sending to PostgreSQL")
            print(f"2. Use parameterized queries instead of string concatenation")
            print(f"3. Apply the principle of least privilege to database users")
            print(f"4. Monitor PostgreSQL logs for suspicious activity{Colors.ENDC}")
        else:
            vulnerability_status = "NOT VULNERABLE"
            print(f"{Colors.BOLD}{Colors.GREEN}[✓] RESULT: {vulnerability_status} - Your PostgreSQL version appears to be patched{Colors.ENDC}")
            
            safe_box = f"""
{Colors.GREEN}╔{'═' * 68}╗
║{' ' * 68}║
║  {Colors.BOLD}✓  SECURE: No Vulnerability Detected  ✓{Colors.ENDC}{Colors.GREEN}{' ' * 20}║
║{' ' * 68}║
╚{'═' * 68}╝{Colors.ENDC}
"""
            print(safe_box)
            
            print(f"{Colors.BLUE}\nNotes:{Colors.ENDC}")
            print(f"{Colors.BLUE}- The PostgreSQL version you are running is not in the affected version range")
            print(f"- However, always keep your database updated with the latest security patches")
            print(f"- Continue following security best practices for your PostgreSQL deployments{Colors.ENDC}")
    
    except psycopg2.Error as e:
        print(f"\n{Colors.RED}Error connecting to PostgreSQL: {e}{Colors.ENDC}")
        
        error_box = f"""
{Colors.RED}╔{'═' * 68}╗
║{' ' * 68}║
║  {Colors.BOLD}⚠️  CONNECTION ERROR  ⚠️{Colors.ENDC}{Colors.RED}{' ' * 39}║
║{' ' * 68}║
╚{'═' * 68}╝{Colors.ENDC}
"""
        print(error_box)
        
        print(f"{Colors.YELLOW}\nCould not complete the vulnerability check. Please verify your connection details.{Colors.ENDC}")
        sys.exit(1)
    
    print(f"\n{Colors.CYAN}Scan completed at: {Colors.BOLD}" + time.strftime("%Y-%m-%d %H:%M:%S") + f"{Colors.ENDC}")

if __name__ == "__main__":
    main()

Usage:
python3 CVE-2025-1094-checker.py --host db.example.com --port 5432 --user dbadmin --dbname production