// Detect multiple WebSocket connections from same source IP
let TimeWindow = 1h;
let ConnectionThreshold = 50;
DeviceNetworkEvents
| where ActionType == "WebSocketConnectionAccepted"
| summarize count() by SrcIpAddr, bin(Timestamp, TimeWindow)
| where count_ > ConnectionThreshold
| project suspicious_ip=SrcIpAddr, connection_count=count_
  

// Detect OCPP connection attempts without proper authentication flow
let OCPPPorts = dynamic([8000, 443, 80]);
DeviceNetworkEvents
| where RemotePort in OCPPPorts
| where ActionType has "WebSocket"
| where isnull(AdditionalFields)
| summarize by SrcIpAddr, DeviceName, Timestamp
| sort by Timestamp desc


// Detect session identifier enumeration patterns
let TimeWindow = 10m;
let EnumThreshold = 100;
DeviceNetworkEvents
| where RemotePort == 443 and ActionType has "WebSocket"
| where NetworkProtocol == "WebSocket" 
| project Timestamp, SrcIpAddr, DstIpAddress, RequestFields
| parse RequestFields with * "stationId":"" StationId ""* 
| where isnotempty(StationId)
| summarize distinct_station_count = dcount(StationId) by SrcIpAddr, bin(Timestamp, TimeWindow)
| where distinct_station_count > EnumThreshold

#!/usr/bin/env python3
"""
OCPP Security Assessment Tool
Tests for common vulnerabilities in OCPP WebSocket implementations
"""

import asyncio
import websockets
import 
import argparse
from datetime import datetime

class OCPPScanner:
    def __init__(self, target_uri, station_ids=None):
        self.target_uri = target_uri
        self.station_ids = station_ids or []
        self.vulnerabilities = []
        
    async def test_missing_auth(self, station_id):
        """Test CVE-2026-26288: Missing Authentication for Critical Function"""
        try:
            uri = f"{self.target_uri}/{station_id}"
            async with websockets.connect(uri) as websocket:
                # Try to send OCPP command without authentication
                message = [
                    2,  # MessageTypeId: CALL
                    "test123",  # UniqueId
                    "BootNotification",  # Action
                    {
                        "chargePointVendor": "Test",
                        "chargePointModel": "VulnerabilityScan"
                    }
                ]
                await websocket.send(.dumps(message))
                response = await asyncio.wait_for(websocket.recv(), timeout=5)
                
                if response:
                    self.vulnerabilities.append({
                        "cve": "CVE-2026-26288",
                        "severity": "CRITICAL",
                        "finding": "Unauthenticated WebSocket connection accepted",
                        "station_id": station_id
                    })
                    return True
        except Exception as e:
            pass
        return False
    
    async def test_rate_limiting(self, station_id):
        """Test CVE-2026-24696: Missing Rate Limiting on Authentication"""
        connection_count = 0
        start_time = datetime.now()
        
        for i in range(20):  # Attempt 20 rapid connections
            try:
                uri = f"{self.target_uri}/{station_id}"
                async with websockets.connect(uri) as websocket:
                    connection_count += 1
                await asyncio.sleep(0.1)
            except:
                pass
        
        elapsed = (datetime.now() - start_time).total_seconds()
        if connection_count > 10 and elapsed < 5:
            self.vulnerabilities.append({
                "cve": "CVE-2026-24696", 
                "severity": "HIGH",
                "finding": f"No rate limiting detected: {connection_count} connections in {elapsed:.2f}s",
                "station_id": station_id
            })
            return True
        return False
    
    async def scan(self):
        tasks = []
        for station_id in self.station_ids:
            tasks.append(self.test_missing_auth(station_id))
            tasks.append(self.test_rate_limiting(station_id))
        
        if not self.station_ids:
            # Test with common patterns if no IDs provided
            test_ids = ["EV-001", "CS-001", "charger-1", "station1"]
            for station_id in test_ids:
                tasks.append(self.test_missing_auth(station_id))
        
        await asyncio.gather(*tasks, return_exceptions=True)
        return self.vulnerabilities

async def main():
    parser = argparse.ArgumentParser(description='OCPP Security Scanner')
    parser.add_argument('target', help='Target WebSocket URI (e.g., ws://api.example.com)')
    parser.add_argument('--station-ids', nargs='+', help='List of station IDs to test')
    parser.add_argument('--output', help='Output JSON file')
    args = parser.parse_args()
    
    scanner = OCPPScanner(args.target, args.station_ids)
    print(f"[*] Starting OCPP security scan against {args.target}")
    
    vulnerabilities = await scanner.scan()
    
    print(f"[*] Scan complete. Found {len(vulnerabilities)} potential vulnerabilities")
    
    for vuln in vulnerabilities:
        print(f"\n[!] {vuln['cve']} ({vuln['severity']})")
        print(f"    {vuln['finding']}")
        if 'station_id' in vuln:
            print(f"    Affected Station ID: {vuln['station_id']}")
    
    if args.output:
        with open(args.output, 'w') as f:
            .dump(vulnerabilities, f, indent=2)
        print(f"[*] Results saved to {args.output}")

if __name__ == "__main__":
    asyncio.run(main())

# Monitor for WebSocket connection spikes to charging infrastructure
#!/bin/bash

THRESHOLD=100
LOG_FILE="/var/log/syslog"
PATTERN="WebSocket.*OCPP"

echo "Monitoring for WebSocket connection anomalies..."

# Count connections in last 5 minutes
connection_count=$(journalctl --since "5 minutes ago" | grep -c "$PATTERN")

if [ "$connection_count" -gt "$THRESHOLD" ]; then
    echo "[ALERT] High WebSocket connection volume detected: $connection_count in 5 minutes"
    # Extract top source IPs
    journalctl --since "5 minutes ago" | grep "$PATTERN" | \
        awk '{print $5}' | sort | uniq -c | sort -rn | head -10
fi


# Network-level detection of OCPP traffic patterns
#!/bin/bash

INTERFACE="eth0"
OCPP_PORTS="8000,443,8080"

# Capture and analyze OCPP traffic
sudo tcpdump -i $INTERFACE -nn port $OCPP_PORTS -c 100 -A 2>/dev/null | \
    grep -E "BootNotification|StartTransaction|StopTransaction" | \
    awk '{print $1, $2, $3, $NF}' > /tmp/ocpp_analysis.log

# Check for suspicious patterns
if grep -q "BootNotification.*guest\|BootNotification.*admin" /tmp/ocpp_analysis.log; then
    echo "[WARNING] Potential weak credentials detected in OCPP traffic"
fi

# Example secure WebSocket configuration
wss_config:
  require_client_cert: true
  client_ca_file: /etc/ssl/certs/charging-stations-ca.pem
  cert_file: /etc/ssl/certs/backend-cert.pem
  key_file: /etc/ssl/private/backend-key.pem
  verify_client: "require"
  cipher_suites:
    - "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384"
    - "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256"


5. **Rate Limiting and Session Management**: Implement application-level rate limiting on all authentication endpoints. Use Redis or similar in-memory stores to track connection attempts per IP and per station identifier.

# Example rate limiting middleware for Python/FastAPI
from fastapi import FastAPI, Request, HTTPException
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()

# OCPP connections: 10 per minute per IP
@app.post("/ocpp/{station_id}")
@limiter.limit("10/minute")
async def ocpp_connection(station_id: str, request: Request):
    # Additional validation for station ID format
    if not validate_station_id(station_id):
        raise HTTPException(status_code=400, detail="Invalid station identifier")
    
    # Check for active session conflicts
    if await has_active_session(station_id):
        raise HTTPException(status_code=409, detail="Station already connected")
    
    return await establish_ocpp_session(station_id)


6. **Telemetry Monitoring**: Deploy continuous monitoring solutions that can detect anomalous charging patterns, unusual command sequences, and geographic impossibilities (e.g., a charger in Texas reporting to a backend in Europe).