Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

MITRE ATT&CK

NIST CSF 2.0

When to Use

When assessing the security posture of HMI systems in SCADA/DCS environments
When evaluating web-based HMI interfaces for common web vulnerabilities
When auditing HMI authentication, authorization, and session management
When testing communication security between HMIs and PLCs/RTUs
When preparing for IEC 62443 or NERC CIP compliance assessments

Do not use for testing HMIs in active production without a maintenance window and rollback plan, for PLC-level protocol analysis (see performing-s7comm-protocol-security-analysis), or for general web application testing on non-OT systems.

Prerequisites

HMI system inventory with vendor, version, and network configuration details
Lab or test environment mirroring production HMI setup (preferred for active testing)
Authorization from plant operations for testing during maintenance windows
NIST SP 800-82 and IEC 62443 security requirements documentation
Network capture capability on HMI-to-PLC communication segment

Workflow

Step 1: Assess HMI Attack Surface

#!/usr/bin/env python3
"""SCADA HMI Security Assessment Tool.
 
Evaluates HMI security across authentication, communication,
configuration, and web interface categories aligned with
IEC 62443 and NIST SP 800-82 requirements.
"""
 
import json
import sys
from datetime import datetime
from typing import Dict, List
 
try:
    import requests
except ImportError:
    print("Install requests: pip install requests")
    sys.exit(1)
 
 
class HMISecurityAssessment:
    """Performs security assessment of SCADA HMI systems."""
 
    def __init__(self, hmi_info: dict):
        self.hmi_info = hmi_info
        self.findings = []
        self.checks_run = 0
        self.checks_passed = 0
 
    def check_authentication(self):
        """Assess HMI authentication mechanisms."""
        checks = [
            {
                "id": "AUTH-01",
                "name": "Password complexity enforcement",
                "iec62443_ref": "ISA-62443-3-3 SR 1.7",
                "description": "HMI must enforce minimum password complexity requirements",
                "test": "Verify minimum length >= 8, complexity rules, history >= 5",
            },
            {
                "id": "AUTH-02",
                "name": "Account lockout policy",
                "iec62443_ref": "ISA-62443-3-3 SR 1.11",
                "description": "HMI must lock accounts after failed login attempts",
                "test": "Verify lockout after 5 failed attempts, lockout duration >= 15 min",
            },
            {
                "id": "AUTH-03",
                "name": "Default credentials changed",
                "iec62443_ref": "ISA-62443-3-3 SR 1.5",
                "description": "All default vendor credentials must be changed",
                "test": "Attempt login with known vendor defaults (admin/admin, operator/operator)",
            },
            {
                "id": "AUTH-04",
                "name": "Role-based access control",
                "iec62443_ref": "ISA-62443-3-3 SR 2.1",
                "description": "HMI must separate operator, engineer, and admin roles",
                "test": "Verify operator role cannot access engineering functions",
            },
            {
                "id": "AUTH-05",
                "name": "Session timeout enforcement",
                "iec62443_ref": "ISA-62443-3-3 SR 1.12",
                "description": "HMI sessions must time out after inactivity",
                "test": "Verify session timeout <= 15 minutes for operator, <= 5 for admin",
            },
            {
                "id": "AUTH-06",
                "name": "Multi-factor authentication for remote access",
                "iec62443_ref": "ISA-62443-3-3 SR 1.13",
                "description": "Remote HMI access requires MFA",
                "test": "Verify MFA is enforced for all non-local HMI connections",
            },
        ]
 
        print(f"\n--- AUTHENTICATION ASSESSMENT ---")
        for check in checks:
            self.checks_run += 1
            print(f"  [{check['id']}] {check['name']}")
            print(f"    Ref: {check['iec62443_ref']}")
            print(f"    Test: {check['test']}")
 
    def check_communication_security(self):
        """Assess HMI-to-PLC communication security."""
        checks = [
            {
                "id": "COMM-01",
                "name": "Encrypted HMI-PLC communication",
                "description": "Traffic between HMI and PLCs should use encrypted protocols (OPC UA with TLS)",
                "test": "Capture HMI-PLC traffic and verify encryption (Wireshark TLS handshake)",
            },
            {
                "id": "COMM-02",
                "name": "HMI write command authentication",
                "description": "Write commands from HMI to PLC should be authenticated",
                "test": "Verify that write operations require operator confirmation/authentication",
            },
            {
                "id": "COMM-03",
                "name": "Web HMI uses HTTPS",
                "description": "Web-based HMI interfaces must use TLS 1.2+ with valid certificates",
                "test": "Check TLS version, cipher suites, certificate validity",
            },
            {
                "id": "COMM-04",
                "name": "No cleartext protocols in use",
                "description": "Telnet, FTP, HTTP must not be used for HMI access or management",
                "test": "Port scan HMI for cleartext protocol services",
            },
        ]
 
        print(f"\n--- COMMUNICATION SECURITY ASSESSMENT ---")
        for check in checks:
            self.checks_run += 1
            print(f"  [{check['id']}] {check['name']}")
            print(f"    Test: {check['test']}")
 
    def check_web_hmi_security(self):
        """Assess web-based HMI for common web vulnerabilities."""
        hmi_url = self.hmi_info.get("url", "")
        if not hmi_url:
            print(f"\n  [SKIP] No web HMI URL provided")
            return
 
        checks = [
            {
                "id": "WEB-01",
                "name": "Cross-Site Scripting (XSS)",
                "owasp": "A7:2017",
                "test": "Test input fields with XSS payloads in tag names, alarm messages",
            },
            {
                "id": "WEB-02",
                "name": "Cross-Site Request Forgery (CSRF)",
                "owasp": "A8:2013",
                "test": "Verify CSRF tokens on state-changing operations (setpoint changes)",
            },
            {
                "id": "WEB-03",
                "name": "Insecure Direct Object References",
                "owasp": "A4:2013",
                "test": "Manipulate URL parameters to access other users HMI views",
            },
            {
                "id": "WEB-04",
                "name": "Security Headers",
                "test": "Verify X-Frame-Options, CSP, X-Content-Type-Options headers",
            },
            {
                "id": "WEB-05",
                "name": "Privileged file system access (CVE-2025-0921)",
                "test": "Check Ignition SCADA for privileged file system vulnerability via project files",
            },
        ]
 
        print(f"\n--- WEB HMI SECURITY ASSESSMENT ---")
        print(f"  Target: {hmi_url}")
        for check in checks:
            self.checks_run += 1
            print(f"  [{check['id']}] {check['name']}")
            print(f"    Test: {check['test']}")
 
    def check_hardening(self):
        """Assess HMI operating system and application hardening."""
        checks = [
            {
                "id": "HARD-01",
                "name": "OS patch level",
                "test": "Verify HMI OS is patched within SLA (typically 90 days for OT)",
            },
            {
                "id": "HARD-02",
                "name": "Unnecessary services disabled",
                "test": "Verify no unnecessary network services running (RDP if not needed, SMB, etc)",
            },
            {
                "id": "HARD-03",
                "name": "USB port restrictions",
                "test": "Verify USB mass storage is blocked on HMI terminals",
            },
            {
                "id": "HARD-04",
                "name": "Application whitelisting",
                "test": "Verify only authorized HMI applications can execute",
            },
            {
                "id": "HARD-05",
                "name": "Audit logging enabled",
                "test": "Verify operator actions, login events, and setpoint changes are logged",
            },
        ]
 
        print(f"\n--- HMI HARDENING ASSESSMENT ---")
        for check in checks:
            self.checks_run += 1
            print(f"  [{check['id']}] {check['name']}")
            print(f"    Test: {check['test']}")
 
    def generate_report(self):
        """Generate assessment report."""
        self.check_authentication()
        self.check_communication_security()
        self.check_web_hmi_security()
        self.check_hardening()
 
        print(f"\n{'='*70}")
        print("SCADA HMI SECURITY ASSESSMENT SUMMARY")
        print(f"{'='*70}")
        print(f"Date: {datetime.now().isoformat()}")
        print(f"HMI: {self.hmi_info.get('name', 'Unknown')}")
        print(f"Vendor: {self.hmi_info.get('vendor', 'Unknown')}")
        print(f"Version: {self.hmi_info.get('version', 'Unknown')}")
        print(f"Total Checks: {self.checks_run}")
        print(f"Findings: {len(self.findings)}")
 
 
if __name__ == "__main__":
    assessment = HMISecurityAssessment(hmi_info={
        "name": "Plant-HMI-01",
        "vendor": "Siemens WinCC",
        "version": "7.5 SP2",
        "ip": "10.10.2.10",
        "url": "https://10.10.2.10:8080",
        "os": "Windows 10 LTSC 2021",
    })
    assessment.generate_report()

Key Concepts

Term	Definition
HMI	Human-Machine Interface providing operators visual representation and control of industrial processes
Web HMI	Browser-based HMI interface accessible via HTTP/HTTPS, subject to standard web vulnerabilities
Setpoint	Target value for a process variable that operators can change through the HMI; unauthorized changes can cause process upset
Alarm Suppression	Attacker technique of disabling or hiding HMI alarms to mask malicious process manipulation
WinCC	Siemens SCADA/HMI software widely deployed in manufacturing and process industries
CVE-2025-0921	Ignition SCADA privileged file system vulnerability exploitable through malicious project uploads

Output Format

HMI SECURITY ASSESSMENT REPORT
=================================
Date: YYYY-MM-DD
HMI: [name] | Vendor: [vendor] | Version: [version]
 
FINDINGS BY CATEGORY:
  Authentication: [pass/fail count]
  Communication: [pass/fail count]
  Web Security: [pass/fail count]
  Hardening: [pass/fail count]
 
CRITICAL FINDINGS:
  1. [finding with remediation]
 
COMPLIANCE STATUS:
  IEC 62443 SL-T: [target level]
  IEC 62443 SL-A: [achieved level]

Source materials

References and resources

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

References 1

api-reference.md2.5 KB

SCADA HMI Security Assessment - API Reference

SCADA Protocol Ports

Port	Protocol	Description
102	S7comm	Siemens S7 PLC communication
502	Modbus TCP	Industrial automation protocol
2222	EtherNet/IP	Allen-Bradley, Rockwell
4840	OPC UA	Open Platform Communications Unified Architecture
20000	DNP3	Distributed Network Protocol
47808	BACnet	Building Automation and Control

Port Scanning (socket stdlib)

import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2.0)
result = sock.connect_ex((target, port))  # 0 = open
sock.close()

pyshark for Protocol Analysis

import pyshark
cap = pyshark.FileCapture("traffic.pcap")
for pkt in cap:
    for layer in pkt.layers:
        print(layer.layer_name)  # modbus, s7comm, dnp3, etc.
cap.close()

Insecure SCADA Protocols

These protocols lack built-in encryption and authentication:

Modbus TCP - No auth, no encryption, commands in plaintext
S7comm - No auth (pre-V4), no encryption
DNP3 - Optional Secure Authentication (SA), rarely deployed
BACnet - No native security mechanisms
EtherNet/IP - No encryption, device enumeration possible

HMI Configuration Checks

Check	Severity	Description
Authentication disabled	Critical	HMI allows anonymous access
No session timeout	High	Sessions persist indefinitely
TLS disabled	High	Communications in plaintext
Remote access without VPN	Critical	HMI exposed without tunnel
No RBAC	High	Single role or no access control
Default credentials	Critical	Factory-default username/password

Common Default Credentials

Username	Password	Platform
admin	admin	Generic HMI
admin	1234	Siemens WinCC
operator	operator	Wonderware
engineer	engineer	GE iFIX
guest	guest	Various

ICS Security Standards

IEC 62443 - Industrial communication network security
NIST SP 800-82 - Guide to ICS Security
NERC CIP - Critical Infrastructure Protection (power grid)

Output Schema

{
  "report": "scada_hmi_security_assessment",
  "target": "192.168.1.100",
  "total_findings": 6,
  "severity_summary": {"critical": 2, "high": 3, "medium": 1},
  "findings": [{"type": "open_scada_port", "severity": "high"}]
}

CLI Usage

python agent.py --target 192.168.1.100 --pcap traffic.pcap --config hmi.json --output report.json

Scripts 1

agent.py6.7 KB

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

#!/usr/bin/env python3
"""SCADA HMI Security Assessment agent — analyzes SCADA HMI configurations
for security weaknesses including default credentials, unencrypted protocols,
and missing access controls."""

import argparse
import json
import socket
from collections import Counter
from datetime import datetime
from pathlib import Path

try:
    import pyshark
except ImportError:
    pyshark = None

SCADA_PORTS = {
    102: "S7comm (Siemens)",
    502: "Modbus TCP",
    2222: "EtherNet/IP",
    4840: "OPC UA",
    20000: "DNP3",
    47808: "BACnet",
    1089: "FF HSE",
    18245: "GE SRTP",
}

DEFAULT_CREDENTIALS = [
    ("admin", "admin"), ("admin", "password"), ("admin", "1234"),
    ("operator", "operator"), ("engineer", "engineer"),
    ("guest", "guest"), ("user", "user"),
    ("Administrator", ""), ("root", "root"),
]

INSECURE_PROTOCOLS = {"modbus", "s7comm", "dnp3", "bacnet", "enip"}


def scan_open_ports(target: str, ports: list[int] = None, timeout: float = 2.0) -> list[dict]:
    """Check for open SCADA-specific ports on target."""
    if ports is None:
        ports = list(SCADA_PORTS.keys())
    results = []
    for port in ports:
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(timeout)
            result = sock.connect_ex((target, port))
            if result == 0:
                results.append({
                    "port": port,
                    "protocol": SCADA_PORTS.get(port, "unknown"),
                    "status": "open",
                    "risk": "high" if port in (502, 102, 20000) else "medium",
                })
            sock.close()
        except socket.error:
            pass
    return results


def check_default_credentials_http(target: str, port: int = 80) -> list[dict]:
    """Check for default credentials on HMI web interface."""
    import urllib.request
    import base64
    findings = []
    for user, pwd in DEFAULT_CREDENTIALS:
        try:
            url = f"http://{target}:{port}/"
            creds = base64.b64encode(f"{user}:{pwd}".encode()).decode()
            req = urllib.request.Request(url, headers={"Authorization": f"Basic {creds}"})
            resp = urllib.request.urlopen(req, timeout=5)
            if resp.status == 200:
                findings.append({
                    "type": "default_credential",
                    "severity": "critical",
                    "username": user,
                    "port": port,
                    "detail": f"Default credential {user}:{pwd} accepted on port {port}",
                })
        except Exception:
            continue
    return findings


def analyze_pcap_protocols(pcap_path: str) -> list[dict]:
    """Analyze PCAP for insecure SCADA protocols."""
    if pyshark is None:
        return [{"error": "pyshark not installed"}]
    findings = []
    protocol_counts = Counter()
    try:
        cap = pyshark.FileCapture(pcap_path)
        for pkt in cap:
            for layer in pkt.layers:
                lname = layer.layer_name.lower()
                if lname in INSECURE_PROTOCOLS:
                    protocol_counts[lname] += 1
        cap.close()
    except Exception as e:
        return [{"error": str(e)}]

    for proto, count in protocol_counts.items():
        findings.append({
            "type": "insecure_protocol",
            "severity": "high",
            "protocol": proto,
            "packet_count": count,
            "detail": f"{proto} traffic detected ({count} packets) — no encryption or authentication",
        })
    return findings


def check_hmi_configuration(config_path: str) -> list[dict]:
    """Analyze HMI configuration file for security weaknesses."""
    findings = []
    try:
        config = json.loads(Path(config_path).read_text(encoding="utf-8"))
    except (json.JSONDecodeError, FileNotFoundError) as e:
        return [{"error": str(e)}]

    if not config.get("authentication", {}).get("enabled", True):
        findings.append({"type": "auth_disabled", "severity": "critical",
                         "detail": "Authentication is disabled on HMI"})
    if config.get("session_timeout", 0) == 0:
        findings.append({"type": "no_session_timeout", "severity": "high",
                         "detail": "No session timeout configured"})
    if not config.get("encryption", {}).get("tls_enabled", True):
        findings.append({"type": "no_tls", "severity": "high",
                         "detail": "TLS not enabled for HMI communications"})
    if config.get("remote_access", {}).get("enabled", False):
        if not config.get("remote_access", {}).get("vpn_required", True):
            findings.append({"type": "remote_no_vpn", "severity": "critical",
                             "detail": "Remote access enabled without VPN requirement"})
    roles = config.get("roles", [])
    if len(roles) <= 1:
        findings.append({"type": "no_rbac", "severity": "high",
                         "detail": "No role-based access control — single role or no roles defined"})
    return findings


def generate_report(target: str, pcap_path: str = None,
                    config_path: str = None, scan_ports: bool = True) -> dict:
    """Run all assessments and build consolidated report."""
    findings = []
    if scan_ports:
        open_ports = scan_open_ports(target)
        for p in open_ports:
            findings.append({"type": "open_scada_port", "severity": p["risk"],
                             "detail": f"Port {p['port']} ({p['protocol']}) is open"})
    if pcap_path:
        findings.extend(analyze_pcap_protocols(pcap_path))
    if config_path:
        findings.extend(check_hmi_configuration(config_path))

    severity_counts = Counter(f.get("severity", "info") for f in findings)
    return {
        "report": "scada_hmi_security_assessment",
        "generated_at": datetime.utcnow().isoformat() + "Z",
        "target": target,
        "total_findings": len(findings),
        "severity_summary": dict(severity_counts),
        "findings": findings,
    }


def main():
    parser = argparse.ArgumentParser(description="SCADA HMI Security Assessment Agent")
    parser.add_argument("--target", required=True, help="Target HMI IP address")
    parser.add_argument("--pcap", help="PCAP file with SCADA traffic")
    parser.add_argument("--config", help="HMI configuration JSON file")
    parser.add_argument("--no-scan", action="store_true", help="Skip port scanning")
    parser.add_argument("--output", help="Output JSON file path")
    args = parser.parse_args()

    report = generate_report(args.target, args.pcap, args.config, not args.no_scan)
    output = json.dumps(report, indent=2)
    if args.output:
        Path(args.output).write_text(output, encoding="utf-8")
        print(f"Report written to {args.output}")
    else:
        print(output)


if __name__ == "__main__":
    main()

Keep exploring