Running SQL over a TCP connection

python3 sql-over-raw-tcp.py

# Description: This script demonstrates how to run a SQL query over
# a raw TCP connection to a PostgreSQL server

# Usage: python3 sql-over-raw-tcp.py
# Date Created: 2025-02-04
# Author: Kshitij Purwar
# Blog: https://therookie.pro/posts/running-sql-over-raw-tcp-connection/

import socket

PSQL_URL = "localhost"
PSQ_PORT = 5432
PSQL_PASSWORD = "mysecretpassword"
ENCODING = "utf-8"


def main():
    psql_startup_message = "\0".join(
        ["user", "postgres", "database", "postgres", "client_encoding", "UTF8"]
    )

    print(psql_startup_message)
    # Each character is represented by a byte
    print(len(psql_startup_message))  # 74

    # create a socket object
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((PSQL_URL, PSQ_PORT))

    # 4 bytes each for the message length and protocol version # 0000005400030000
    # 2 bytes for extra null bytes for the whole message i.e. at the end \0\0
    # Total bytes = 4 + 4 + 74 + 2 = 84 i.e. T

    MESSAGE_PROTOCOL_MAJOR_VERSION = 3
    MESSAGE_PROTOCOL_MINOR_VERSION = 0
    complete_startup_message = (
        (len(psql_startup_message) + 10).to_bytes(4, "big")
        + MESSAGE_PROTOCOL_MAJOR_VERSION.to_bytes(2, "big")
        + MESSAGE_PROTOCOL_MINOR_VERSION.to_bytes(2, "big")
        + psql_startup_message.encode(ENCODING)
        + b"\0\0"
    )
    pretty_print_hex(complete_startup_message)

    sock.sendall(complete_startup_message)
    response = sock.recv(1024)  # 1024 is the buffer size
    divider()
    pretty_print_hex(response)

    # 1 byte for the PostgreSQL protocol message type indicator i.e. p or Q
    # 4 bytes for the message length

    # 0000   70 00 00 00 15 6d 79 73 65 63 72 65 74 70 61 73   p....mysecretpas
    # 0010   73 77 6f 72 64 00                                 sword.
    password_message = (
        b"p"
        + (len(PSQL_PASSWORD) + 5).to_bytes(4, "big")
        + PSQL_PASSWORD.encode(ENCODING)
        + b"\0"
    )
    pretty_print_hex(password_message)

    sock.sendall(password_message)
    response = sock.recv(1024)
    divider()
    pretty_print_hex(response)

    # 0000   51 00 00 00 10 53 45 4c 45 43 54 20 32 2b 35 3b   Q....SELECT 2+5;
    # 0010   00                                                .
    SQL_QUERY = "SELECT 2+5;"

    query_message = (
        b"Q"
        + (len(SQL_QUERY) + 5).to_bytes(4, "big")
        + SQL_QUERY.encode(ENCODING)
        + b"\0"
    )
    pretty_print_hex(query_message)

    sock.sendall(query_message)
    response = sock.recv(1024)
    divider()
    pretty_print_hex(response)

    termination_message = b"X" + (4).to_bytes(4, "big")
    pretty_print_hex(termination_message)

    sock.sendall(termination_message)
    response = sock.recv(1024)
    divider()
    pretty_print_hex(response)
    
    sock.close()


def pretty_print_hex(hex_string: bytes):
    """Pretty print the hex string
    Args
        hex_string (bytes): The hex string to print

    Returns
        None
    """

    print("\x1b[6;30;42m" + "Binary :" + "\x1b[0m", hex_string)
    # print("")
    print(
        "\x1b[6;30;43m" + "ASCII  :" + "\x1b[0m",
        hex_string.decode(ENCODING, errors="replace"),
    )
    print("\n")


def divider():
    """Print a colored divider"""
    print("\x1b[6;30;41m" + "---- SQL Response ----" + "\x1b[0m")


if __name__ == "__main__":
    main()