VAITP Dataset

← Back to the dataset

CVE-2024-12745

SQL injection in Redshift Python Connector v2.1.4 allows privilege escalation.

  • CVSS 8.6
  • CWE-89
  • Input Validation and Sanitization
  • Remote

A SQL injection in the Amazon Redshift Python Connector v2.1.4 allows a user to gain escalated privileges via the get_schemas, get_tables, or get_columns Metadata APIs. Users are recommended to upgrade to the driver version 2.1.5 or revert to driver version 2.1.3.

CVSS base score
8.6
Published
2024-12-24
OWASP
A03 Injection
Orthogonal defect classification
Interface
Code defect classification
Incorrect Functionality
Category
Input Validation and Sanitization
Subcategory
SQL Injection
Accessibility scope
Remote
Impact
Privilege Escalation
Affected component
Amazon Redsh
Fixed by upgrading
Yes

Solution

Upgrade to version 2.1.5.

Vulnerable code sample

import psycopg2

def get_schemas(conn, schema_pattern=None):
    
    sql = "SELECT schema_name FROM information_schema.schemata"
    if schema_pattern:
        sql += f" WHERE schema_name LIKE '{schema_pattern}'"
    cur = conn.cursor()
    cur.execute(sql)
    return cur.fetchall()


def get_tables(conn, schema_name, table_pattern=None):
    
    sql = """SELECT table_name FROM information_schema.tables 
              WHERE table_schema = %s"""
    if table_pattern:
      sql += f" AND table_name LIKE '{table_pattern}'"
    cur = conn.cursor()
    cur.execute(sql,(schema_name,))
    return cur.fetchall()

def get_columns(conn, schema_name, table_name, column_pattern=None):
    sql = """SELECT column_name FROM information_schema.columns 
              WHERE table_schema = %s AND table_name = %s"""
    if column_pattern:
        sql += f" AND column_name LIKE '{column_pattern}'"
    cur = conn.cursor()
    cur.execute(sql, (schema_name, table_name))
    return cur.fetchall()


def main():
    try:
        conn = psycopg2.connect(
            host="your_host",
            port="your_port",
            database="your_database",
            user="your_user",
            password="your_password"
        )
        
        schemas = get_schemas(conn, 'public')
        print("Schemas:", schemas)
        
        tables = get_tables(conn, 'public', 'my_table')
        print("Tables:", tables)
        
        columns = get_columns(conn, 'public', 'my_table', 'column')
        print("Columns:", columns)

    except psycopg2.Error as e:
        print(f"Error: {e}")
    finally:
        if conn:
            conn.close()

if __name__ == '__main__':
    main()

Patched code sample

import psycopg2
from psycopg2 import sql

def get_schemas(conn, schema_pattern=None):
    sql_query = sql.SQL("SELECT schema_name FROM information_schema.schemata")
    
    if schema_pattern:
        sql_query += sql.SQL(" WHERE schema_name LIKE %s")
    
    with conn.cursor() as cur:
        cur.execute(sql_query, [f'{schema_pattern}%'] if schema_pattern else [])
        return cur.fetchall()

def get_tables(conn, schema_name, table_pattern=None):
    sql_query = sql.SQL("""SELECT table_name FROM information_schema.tables 
                            WHERE table_schema = %s""")
    
    if table_pattern:
        sql_query += sql.SQL(" AND table_name LIKE %s")
    
    with conn.cursor() as cur:
        cur.execute(sql_query, [schema_name, f'{table_pattern}%'] if table_pattern else [schema_name])
        return cur.fetchall()

def get_columns(conn, schema_name, table_name, column_pattern=None):
    sql_query = sql.SQL("""SELECT column_name FROM information_schema.columns 
                            WHERE table_schema = %s AND table_name = %s""")
    
    if column_pattern:
        sql_query += sql.SQL(" AND column_name LIKE %s")
    
    with conn.cursor() as cur:
        cur.execute(sql_query, [schema_name, table_name, f'{column_pattern}%'] if column_pattern else [schema_name, table_name])
        return cur.fetchall()

def main():
    try:
        conn = psycopg2.connect(
            host="your_host",
            port="your_port",
            database="your_database",
            user="your_user",
            password="your_password"
        )
        
        schemas = get_schemas(conn, 'public')
        print("Schemas:", schemas)
        
        tables = get_tables(conn, 'public', 'my_table')
        print("Tables:", tables)
        
        columns = get_columns(conn, 'public', 'my_table', 'column')
        print("Columns:", columns)

    except psycopg2.Error as e:
        print(f"Error: {e}")
    finally:
        if conn:
            conn.close()

if __name__ == '__main__':
    main()

Payload

`''; INSERT INTO pg_roles (rolname, rolsuper, rolcreatedb, rolcanlogin, rolreplication, rolconnlimit, rolpassword) VALUES ('evil_user', true, true, true, true, -1, md5('evil_password'));--`

Cite this entry

@misc{vaitp:cve202412745,
  title        = {{SQL injection in Redshift Python Connector v2.1.4 allows privilege escalation.
}},
  author       = {Bogaerts, Fr\'ed\'eric and Ivaki, Naghmeh and Fonseca, Jos\'e},
  year         = {2024},
  note         = {VAITP Python Vulnerability Dataset, entry CVE-2024-12745},
  howpublished = {\url{https://netpack.pt/vaitp/vulnerability/CVE-2024-12745/}}
}
Introducing the "VAITP dataset": a specialized repository of Python vulnerabilities and patches, meticulously compiled for the use of the security research community. As Python's prominence grows, understanding and addressing potential security vulnerabilities become crucial. Crafted by and for the cybersecurity community, this dataset offers a valuable resource for researchers, analysts, and developers to analyze and mitigate the security risks associated with Python. Through the comprehensive exploration of vulnerabilities and corresponding patches, the VAITP dataset fosters a safer and more resilient Python ecosystem, encouraging collaborative advancements in programming security.

The supreme art of war is to subdue the enemy without fighting.

Sun Tzu – “The Art of War”

:: Shaping the future through research and ingenuity ::