SCADA/ICS + AI 攻擊

# Historian 層級資料投毒示意
# 目標：工業泵浦的預測性維護模型
def poison_historian_data(historian_connection, target_sensor,
                          poison_strategy):
    """
    Inject manipulated readings into the historian to
    affect AI model training or inference.
    WARNING: Test environment only.
    """
    if poison_strategy == "gradual_drift":
        # Slowly shift readings to desensitize the AI
        for hour in range(720):  # 30 days
            original = historian_connection.read_latest(target_sensor)
            drift_amount = 0.001 * hour
            poisoned_value = original + drift_amount
            historian_connection.inject(
                sensor=target_sensor,
                value=poisoned_value,
                timestamp=current_time(),
            )
    elif poison_strategy == "noise_injection":
        import random
        for reading in historian_connection.stream(target_sensor):
            noise = random.gauss(0, reading.value * 0.05)
            historian_connection.modify(
                reading_id=reading.id,
                new_value=reading.value + noise,
            )
    elif poison_strategy == "label_flipping":
        events = historian_connection.get_labeled_events(target_sensor)
        flip_count = int(len(events) * 0.1)
        for event in random.sample(events, flip_count):
            new_label = "normal" if event.label == "fault" else "fault"
            historian_connection.update_label(event.id, new_label)

# Detection capability assessment
detection_tests = {
    "physics_consistency": {
        "description": "Does the AI cross-reference sensor readings "
                       "against physical laws?",
        "test": "Inject readings that are physically impossible",
        "expected_defense": "AI flags physically inconsistent readings",
    },
    "cross_sensor_validation": {
        "description": "Does the AI validate readings against "
                       "correlated sensors?",
        "test": "Poison one temperature sensor while leaving "
                "adjacent sensors unmodified",
        "expected_defense": "AI detects disagreement between "
                            "correlated sensors",
    },
    "rate_of_change_limits": {
        "description": "Does the AI enforce physically plausible "
                       "rates of change?",
        "test": "Inject a sudden step change in a sensor that "
                "should only change gradually",
        "expected_defense": "AI flags impossible rate of change",
    },
}

# Predictive maintenance manipulation attack chain
maintenance_attack = {
    "phase_1_baseline": {
        "description": "Observe normal maintenance prediction patterns",
        "duration": "2-4 weeks",
        "objective": "Understand which sensor readings trigger "
                     "maintenance alerts and their thresholds",
    },
    "phase_2_desensitization": {
        "description": "Gradually shift 'normal' readings toward "
                       "fault thresholds to desensitize the model",
        "duration": "4-8 weeks",
        "technique": "Small daily increases in vibration or temperature "
                     "readings that the model incorporates as the new "
                     "normal baseline",
    },
    "phase_3_masking": {
        "description": "When real fault indicators appear, the model "
                       "does not recognize them because its baseline "
                       "has been shifted",
        "outcome": "Equipment continues operating past its safe "
                   "maintenance window",
    },
    "phase_4_failure": {
        "description": "Equipment fails because maintenance was not "
                       "performed when needed",
        "impact": "Unplanned downtime, potential safety incident, "
                  "cascading failures in interconnected systems",
    },
}

# Testing HMI AI overlay for manipulation
def test_hmi_ai_overlay(hmi_system, scenarios):
    """
    Present the HMI AI with scenarios that should trigger
    specific operator alerts and verify correct behavior.
    """
    results = []
    for scenario in scenarios:
        hmi_system.inject_readings(scenario.sensor_data)
        overlay_output = hmi_system.get_overlay_display()
        result = {
            "scenario": scenario.name,
            "ground_truth_status": scenario.actual_status,
            "ai_displayed_status": overlay_output.status,
            "correct": overlay_output.status == scenario.actual_status,
            "alerts_shown": overlay_output.active_alerts,
            "expected_alerts": scenario.expected_alerts,
            "alerts_correct": set(overlay_output.active_alerts) ==
                              set(scenario.expected_alerts),
            "suggested_actions": overlay_output.suggested_actions,
            "suggested_actions_safe": all(
                action in scenario.safe_actions
                for action in overlay_output.suggested_actions
            ),
        }
        results.append(result)
    return results