"""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."""
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
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
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())
