npx skills add mukul975/Anthropic-Cybersecurity-SkillsMITRE ATT&CK
When to Use
- When conducting periodic cybersecurity assessments of power grid facilities per NERC CIP requirements
- When assessing substation automation systems using IEC 61850 GOOSE and MMS protocols
- When evaluating the security of an Energy Management System (EMS) or SCADA control center
- When assessing synchrophasor (PMU) networks and wide-area monitoring systems
- When preparing for regional entity compliance audits or internal security reviews
Do not use for non-BES systems below NERC registration thresholds, for general OT assessment without power grid specifics (see performing-ot-network-security-assessment), or for physical security assessment of generation facilities without cyber scope.
Prerequisites
- Understanding of electric power grid architecture (generation, transmission, distribution)
- Familiarity with NERC CIP standards and BES Cyber System categorization
- Knowledge of power grid protocols (IEC 61850, IEC 60870-5-104, DNP3, ICCP/TASE.2)
- Passive monitoring tools for substation network traffic analysis
- Access to EMS/SCADA architecture documentation and network diagrams
Workflow
Step 1: Map Power Grid Cyber Architecture
Identify and document all cyber systems supporting grid operations including EMS, SCADA, substation automation, and communication infrastructure.
# Power Grid Cyber Architecture Assessment
facility_type: "Regional Transmission Organization Control Center"
ems_systems:
primary_ems:
vendor: "GE Grid Solutions"
product: "EMS/SCADA (formerly XA/21)"
functions:
- "State estimation"
- "Automatic generation control (AGC)"
- "Security-constrained economic dispatch"
- "Contingency analysis"
protocols:
- "ICCP/TASE.2 (inter-control center)"
- "DNP3 (substation RTU polling)"
- "IEC 60870-5-104 (substation polling)"
backup_control_center:
location: "Geographically diverse backup site"
sync_method: "Real-time database mirroring"
switchover_time: "< 5 minutes"
substation_automation:
count: 145
system_types:
- vendor: "ABB"
product: "RTU560"
protocol: "DNP3 over TCP/IP"
count: 85
- vendor: "SEL"
product: "SEL-3530 RTAC"
protocol: "IEC 61850 MMS + GOOSE"
count: 40
- vendor: "Siemens"
product: "SICAM A8000"
protocol: "IEC 60870-5-104"
count: 20
communications:
primary: "MPLS WAN (carrier-provided)"
backup: "Licensed microwave radio"
last_mile: "Fiber optic to substation"
synchrophasor_network:
pmu_count: 75
pdc: "GE PDC (Phasor Data Concentrator)"
communication: "IEEE C37.118.2 over dedicated network"
data_rate: "30-60 samples per second"Step 2: Assess Substation Automation Security
Evaluate IEC 61850-based substation automation for protocol security, access controls, and network segmentation.
#!/usr/bin/env python3
"""Power Grid Substation Security Assessor.
Evaluates security of IEC 61850-based substation automation
systems including GOOSE messaging, MMS client/server, and
network architecture.
"""
import json
import sys
from dataclasses import dataclass, field, asdict
from datetime import datetime
@dataclass
class SubstationFinding:
finding_id: str
severity: str
category: str
title: str
description: str
affected_systems: list
nerc_cip_ref: str
iec_62351_ref: str
remediation: str
class SubstationAssessment:
"""Assesses cybersecurity of substation automation systems."""
def __init__(self, substation_name):
self.name = substation_name
self.findings = []
self.counter = 1
def assess_iec61850_security(self, config):
"""Assess IEC 61850 protocol security."""
# GOOSE message authentication
if not config.get("goose_authentication"):
self.findings.append(SubstationFinding(
finding_id=f"SUB-{self.counter:03d}",
severity="critical",
category="Protocol Security",
title="IEC 61850 GOOSE Messages Lack Authentication",
description=(
"GOOSE messages used for protection signaling between IEDs "
"are not authenticated. An attacker on the station bus could "
"inject false trip/close commands to circuit breakers."
),
affected_systems=config.get("goose_publishers", []),
nerc_cip_ref="CIP-005-7 R1.5 - ESP internal communications",
iec_62351_ref="IEC 62351-6 - GOOSE/SV authentication",
remediation=(
"Implement IEC 62351-6 GOOSE authentication using digital "
"signatures. Deploy VLAN isolation for GOOSE traffic as interim."
),
))
self.counter += 1
# MMS service access control
if not config.get("mms_authentication"):
self.findings.append(SubstationFinding(
finding_id=f"SUB-{self.counter:03d}",
severity="high",
category="Protocol Security",
title="MMS Client Connections Lack Authentication",
description=(
"MMS (Manufacturing Message Specification) connections to IEDs "
"do not require client authentication. Any device on the station "
"bus can read/write IED configuration and operate breakers."
),
affected_systems=config.get("mms_servers", []),
nerc_cip_ref="CIP-007-6 R5 - System Access Controls",
iec_62351_ref="IEC 62351-4 - MMS security profiles",
remediation="Enable TLS for MMS connections per IEC 62351-4.",
))
self.counter += 1
# Station bus segmentation
if not config.get("station_bus_segmented"):
self.findings.append(SubstationFinding(
finding_id=f"SUB-{self.counter:03d}",
severity="high",
category="Network Architecture",
title="Flat Station Bus Network Without Segmentation",
description=(
"Station bus connects all IEDs, HMI, engineering access, "
"and WAN gateway on a single VLAN without segmentation."
),
affected_systems=["All station bus devices"],
nerc_cip_ref="CIP-005-7 R1 - ESP boundary",
iec_62351_ref="IEC 62351-10 - Security architecture",
remediation=(
"Segment station bus into VLANs: protection IEDs, "
"measurement IEDs, station HMI, and WAN gateway."
),
))
self.counter += 1
def assess_remote_access(self, config):
"""Assess remote access security for substations."""
if config.get("direct_vendor_access"):
self.findings.append(SubstationFinding(
finding_id=f"SUB-{self.counter:03d}",
severity="critical",
category="Remote Access",
title="Direct Vendor Remote Access to Substation Without MFA",
description=(
"Vendor support has direct VPN access to substation network "
"without traversing an intermediate system or requiring MFA."
),
affected_systems=["Substation WAN gateway"],
nerc_cip_ref="CIP-005-7 R2 - Remote Access Management",
iec_62351_ref="IEC 62351-8 - Role-based access control",
remediation=(
"Route vendor access through corporate jump server with MFA. "
"Implement session recording per CIP-005-7 R2.4."
),
))
self.counter += 1
def generate_report(self):
"""Generate substation assessment report."""
report = []
report.append("=" * 70)
report.append(f"SUBSTATION CYBERSECURITY ASSESSMENT: {self.name}")
report.append(f"Date: {datetime.now().isoformat()}")
report.append("=" * 70)
for sev in ["critical", "high", "medium", "low"]:
findings = [f for f in self.findings if f.severity == sev]
if findings:
report.append(f"\n--- {sev.upper()} ({len(findings)}) ---")
for f in findings:
report.append(f" [{f.finding_id}] {f.title}")
report.append(f" {f.description[:100]}...")
report.append(f" NERC CIP: {f.nerc_cip_ref}")
report.append(f" Remediation: {f.remediation[:80]}...")
return "\n".join(report)
if __name__ == "__main__":
assessment = SubstationAssessment("Substation Alpha - 345kV")
assessment.assess_iec61850_security({
"goose_authentication": False,
"mms_authentication": False,
"station_bus_segmented": False,
"goose_publishers": ["SEL-411L-01", "SEL-411L-02", "SEL-487E-01"],
"mms_servers": ["SEL-3530-RTAC", "ABB-REF615-01"],
})
assessment.assess_remote_access({
"direct_vendor_access": True,
})
print(assessment.generate_report())Key Concepts
| Term | Definition |
|---|---|
| IEC 61850 | International standard for communication networks and systems in substations, using GOOSE for protection signaling and MMS for SCADA data |
| GOOSE | Generic Object Oriented Substation Event - multicast protocol for fast peer-to-peer protection signaling between IEDs (< 4ms trip time) |
| MMS | Manufacturing Message Specification - client/server protocol for reading/writing IED data and operating circuit breakers |
| IEC 62351 | Security standard series for power system communication protocols providing authentication and encryption for IEC 61850, DNP3, and IEC 104 |
| ICCP/TASE.2 | Inter-Control Center Communications Protocol for data exchange between control centers of different utilities |
| Synchrophasor (PMU) | Phasor Measurement Unit providing time-synchronized voltage/current measurements at 30-60 samples/second for wide-area monitoring |
Tools & Systems
- Dragos Platform: OT security platform with specific threat intelligence on power grid-targeting groups (ELECTRUM, KAMACITE)
- SEL-3620 Ethernet Security Gateway: Substation security device providing encryption, access control, and intrusion detection
- GRIDsure: Power grid cybersecurity assessment framework by Idaho National Laboratory
- Wireshark with IEC 61850 Dissector: Protocol analysis for GOOSE and MMS traffic in substations
Output Format
Power Grid Cybersecurity Assessment Report
=============================================
Facility: [Name and Type]
NERC Registration: [Entity ID]
BES Impact Rating: [High/Medium/Low]
SUBSTATION FINDINGS: [N]
EMS/SCADA FINDINGS: [N]
COMMUNICATION FINDINGS: [N]
NERC CIP COMPLIANCE:
CIP-002: [Status]
CIP-005: [Status]
CIP-007: [Status]References and resources
Everything below is rendered for inspection. Script files are read-only and never run.
References 1
api-reference.md1.4 KB
API Reference — Performing Power Grid Cybersecurity Assessment
Libraries Used
- csv: Parse and generate NERC CIP assessment CSV files
CLI Interface
python agent.py template [--output nerc_cip_template.csv]
python agent.py assess --csv completed_assessment.csv
python agent.py esp --firewall-csv esp_rules.csvCore Functions
generate_assessment_template(output_file) — Create NERC CIP checklist
Generates CSV with all requirements from 11 CIP standards.
assess_compliance(assessment_csv) — Score compliance per standard
Calculates pass/fail rates per CIP standard. Lists all gap findings.
assess_esp_security(firewall_csv) — Electronic Security Perimeter audit
Checks for: allow-from-any rules, allow-any-protocol, missing default-deny.
NERC CIP Standards Covered (11)
| Standard | Title | Checks |
|---|---|---|
| CIP-002 | BES Cyber System Categorization | 3 |
| CIP-003 | Security Management Controls | 3 |
| CIP-004 | Personnel & Training | 3 |
| CIP-005 | Electronic Security Perimeter | 4 |
| CIP-006 | Physical Security | 3 |
| CIP-007 | System Security Management | 4 |
| CIP-008 | Incident Reporting | 3 |
| CIP-009 | Recovery Plans | 3 |
| CIP-010 | Configuration Change Management | 3 |
| CIP-011 | Information Protection | 3 |
| CIP-013 | Supply Chain Risk Management | 3 |
Dependencies
No external packages — Python standard library only.
Scripts 1
agent.py7.4 KB
#!/usr/bin/env python3
"""Agent for performing power grid cybersecurity assessment based on NERC CIP standards."""
import json
import argparse
import csv
NERC_CIP_STANDARDS = {
"CIP-002": {"title": "BES Cyber System Categorization", "checks": [
"All BES cyber systems identified and categorized",
"Impact ratings (High/Medium/Low) assigned to all assets",
"Annual review of BES cyber system list completed",
]},
"CIP-003": {"title": "Security Management Controls", "checks": [
"Cybersecurity policy documented and approved",
"Senior manager designated for CIP compliance",
"Annual policy review completed",
]},
"CIP-004": {"title": "Personnel & Training", "checks": [
"Security awareness training completed for all personnel",
"Background checks on personnel with authorized access",
"Access revocation within 24 hours of termination",
]},
"CIP-005": {"title": "Electronic Security Perimeter", "checks": [
"Electronic Security Perimeter (ESP) defined",
"All external routable connectivity through EAP",
"Inbound/outbound access permissions documented",
"Interactive remote access uses encryption and MFA",
]},
"CIP-006": {"title": "Physical Security", "checks": [
"Physical Security Perimeter (PSP) defined",
"Physical access control systems operational",
"Visitor escort and logging procedures in place",
]},
"CIP-007": {"title": "System Security Management", "checks": [
"Ports and services documentation current",
"Security patches evaluated within 35 days",
"Malicious code prevention deployed",
"Security event monitoring enabled",
]},
"CIP-008": {"title": "Incident Reporting", "checks": [
"Cyber security incident response plan documented",
"Incident response plan tested annually",
"Reportable incidents notified to E-ISAC within 1 hour",
]},
"CIP-009": {"title": "Recovery Plans", "checks": [
"Recovery plans for BES cyber systems documented",
"Recovery plans tested annually",
"Backup media stored securely",
]},
"CIP-010": {"title": "Configuration Change Management", "checks": [
"Baseline configurations documented",
"Change management process for cyber systems",
"Vulnerability assessments performed at least every 15 months",
]},
"CIP-011": {"title": "Information Protection", "checks": [
"BES Cyber System Information (BCSI) identified",
"BCSI storage locations protected",
"BCSI disposal procedures documented",
]},
"CIP-013": {"title": "Supply Chain Risk Management", "checks": [
"Supply chain risk management plan documented",
"Vendor risk assessments performed",
"Software integrity verification procedures",
]},
}
def generate_assessment_template(output_file=None):
"""Generate NERC CIP compliance assessment template."""
rows = []
for std_id, std in NERC_CIP_STANDARDS.items():
for i, check in enumerate(std["checks"], 1):
rows.append({
"standard": std_id, "title": std["title"],
"check_id": f"{std_id}-{i:02d}", "requirement": check,
"status": "", "evidence": "", "gap": "", "remediation": "",
})
if output_file:
with open(output_file, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
writer.writeheader()
writer.writerows(rows)
return {"total_checks": len(rows), "standards": list(NERC_CIP_STANDARDS.keys()), "output": output_file}
def assess_compliance(assessment_csv):
"""Score NERC CIP compliance from completed assessment."""
with open(assessment_csv, "r", encoding="utf-8", errors="replace") as f:
reader = csv.DictReader(f)
rows = list(reader)
by_standard = {}
total_pass = 0
total_fail = 0
for row in rows:
std = row.get("standard", "")
status = row.get("status", "").lower()
is_pass = status in ("pass", "compliant", "yes", "met")
is_fail = status in ("fail", "non-compliant", "no", "not met", "gap")
by_standard.setdefault(std, {"pass": 0, "fail": 0, "total": 0})
by_standard[std]["total"] += 1
if is_pass:
by_standard[std]["pass"] += 1
total_pass += 1
elif is_fail:
by_standard[std]["fail"] += 1
total_fail += 1
for std in by_standard:
d = by_standard[std]
d["compliance_pct"] = round(d["pass"] / max(d["total"], 1) * 100, 1)
gaps = [row for row in rows if row.get("status", "").lower() in ("fail", "non-compliant", "no", "not met", "gap")]
return {
"total_requirements": len(rows), "passed": total_pass, "failed": total_fail,
"overall_compliance": round(total_pass / max(len(rows), 1) * 100, 1),
"by_standard": by_standard,
"gaps": [{"standard": g.get("standard"), "check": g.get("check_id"),
"requirement": g.get("requirement", "")[:150],
"gap": g.get("gap", "")[:150]} for g in gaps[:20]],
}
def assess_esp_security(firewall_csv):
"""Assess Electronic Security Perimeter configuration."""
with open(firewall_csv, "r", encoding="utf-8", errors="replace") as f:
reader = csv.DictReader(f)
rules = list(reader)
findings = []
for rule in rules:
action = rule.get("action", "").lower()
src = rule.get("source", rule.get("src", ""))
dst = rule.get("destination", rule.get("dst", ""))
if action == "allow" and src.lower() in ("any", "0.0.0.0/0"):
findings.append({"rule": rule.get("id", rule.get("name", "")), "issue": "ALLOW_FROM_ANY", "severity": "CRITICAL"})
if action == "allow" and rule.get("protocol", "").lower() in ("any", "all"):
findings.append({"rule": rule.get("id", ""), "issue": "ALLOW_ANY_PROTOCOL", "severity": "HIGH"})
has_default_deny = any(r.get("action", "").lower() in ("deny", "drop") and r.get("source", "").lower() in ("any", "0.0.0.0/0")
for r in rules)
if not has_default_deny:
findings.append({"issue": "NO_DEFAULT_DENY", "severity": "CRITICAL"})
return {
"total_rules": len(rules),
"findings": findings[:20],
"default_deny": has_default_deny,
"esp_compliance": "PASS" if not findings else "FAIL",
}
def main():
parser = argparse.ArgumentParser(description="Power Grid Cybersecurity Assessment Agent (NERC CIP)")
sub = parser.add_subparsers(dest="command")
t = sub.add_parser("template", help="Generate NERC CIP assessment template")
t.add_argument("--output", help="Output CSV file")
a = sub.add_parser("assess", help="Score compliance assessment")
a.add_argument("--csv", required=True)
e = sub.add_parser("esp", help="Assess Electronic Security Perimeter")
e.add_argument("--firewall-csv", required=True)
args = parser.parse_args()
if args.command == "template":
result = generate_assessment_template(args.output)
elif args.command == "assess":
result = assess_compliance(args.csv)
elif args.command == "esp":
result = assess_esp_security(args.firewall_csv)
else:
parser.print_help()
return
print(json.dumps(result, indent=2, default=str))
if __name__ == "__main__":
main()