Init commit

ChecksumTools/crcfinder.py

@@ -0,0 +1,36 @@
+import sys
+import argparse
+from typing import List, Tuple
+
+
+def crc(data: List[int], len_: int, start_val: int, poly: int) -> int:
+    crc_val = start_val
+    for i in range(len_):
+        crc_val ^= data[i]
+        for _ in range(8):
+            if (crc_val & 0x80) != 0:
+                crc_val = ((crc_val << 1) ^ poly) & 0xFF
+            else:
+                crc_val <<= 1
+                crc_val &= 0xFF
+    return crc_val ^ 0xFF
+
+def main(data: List[List[int]]):
+    for poly in range(0x01, 0xFF):
+        for start in range(0x00, 0x100):
+            found = 0
+            for pkt in data:
+                if crc(pkt[:7], 7, start, poly) == pkt[7]:
+                    found += 1
+            if found >= 4:
+                print(f"found = {found}! poly = {poly:02X} start = {start:02X}")
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="CRC finder")
+    parser.add_argument("data", nargs="+", help="Data packets (8-byte hex strings)")
+
+    args = parser.parse_args()
+
+    data = [list(bytes.fromhex(pkt)) for pkt in args.data]
+
+    main(data)
+#thank you Alexey Esaulenko for original program design in CAN decoders

ChecksumTools/fuckbmw.py

@@ -0,0 +1,8 @@
+import sys
+
+input_bytes = bytearray.fromhex(sys.argv[1])
+checksum = bytearray.fromhex(sys.argv[2])
+gayness = sum(input_bytes)
+gayness += int.from_bytes(checksum, byteorder='big')
+result = (gayness >> 8) + (gayness & 0xFF)
+print(hex(result))

ChecksumTools/fuckbmw2unique.py

@@ -0,0 +1,19 @@
+import csv
+
+unique_addresses = set()
+with open('input.csv', 'r') as in_file, open('output.csv', 'w', newline='') as out_file:
+
+    csv_reader = csv.DictReader(in_file)
+    csv_writer = csv.writer(out_file)
+    csv_writer.writerow(['Time Stamp', 'ID', 'Extended', 'Dir', 'Bus', 'LEN', 'D1', 'D2', 'D3', 'D4', 'D5', 'D6', 'D7', 'D8'])
+    for row in csv_reader:
+        d_values = [int(row[f"D{i}"], 16) for i in range(2, 9)]
+   
+        gaysome = sum(d_values) + int(row['ID'], 16)
+        gaysome = (gaysome & 0xFF) + (gaysome >> 8)
+
+        if gaysome == int(row['D1'], 16):
+            address = row['ID']
+            if address not in unique_addresses:
+                csv_writer.writerow([row['Time Stamp'], row['ID'], row['Extended'], row['Dir'], row['Bus'], row['LEN'], row['D1'], row['D2'], row['D3'], row['D4'], row['D5'], row['D6'], row['D7'], row['D8']])
+                unique_addresses.add(address)

ChecksumTools/fuckbmw2uniqueCounter.py

@@ -0,0 +1,54 @@
+import csv
+failed_rows = []
+unique_addresses = {}
+
+with open('input.csv', 'r') as in_file, open('output.csv', 'w', newline='') as out_file, open('failedoutputs.csv', 'w', newline='') as fail_file:
+
+
+    csv_reader = csv.DictReader(in_file)
+    csv_writer = csv.writer(out_file)
+    fail_writer = csv.writer(fail_file)
+
+
+    csv_writer.writerow(['Time Stamp', 'ID', 'Extended', 'Dir', 'Bus', 'LEN', 'D1', 'D2', 'D3', 'D4', 'D5', 'D6', 'D7', 'D8'])
+    fail_writer.writerow(['Time Stamp', 'ID', 'Extended', 'Dir', 'Bus', 'LEN', 'D1', 'D2', 'D3', 'D4', 'D5', 'D6', 'D7', 'D8'])
+
+    for row in csv_reader:
+        d_values = [int(row[f"D{i}"], 16) for i in range(2, 9)]
+        checksum = sum(d_values) + int(row['ID'], 16)
+        checksum = (checksum & 0xFF) + (checksum >> 8)
+        d1 = int(row['D1'], 16)
+        if d1 <= checksum or (d1 > checksum and d1 - checksum == 256):
+            address = row['ID']
+            if address not in unique_addresses:
+
+                csv_writer.writerow([row['Time Stamp'], row['ID'], row['Extended'], row['Dir'], row['Bus'], row['LEN'], row['D1'], row['D2'], row['D3'], row['D4'], row['D5'], row['D6'], row['D7'], row['D8']])
+                unique_addresses[address] = [1, 0]
+            else:
+                unique_addresses[address][0] += 1
+        else:
+            failed_rows.append(row)
+            fail_writer.writerow([row['Time Stamp'], row['ID'], row['Extended'], row['Dir'], row['Bus'], row['LEN'], row['D1'], row['D2'], row['D3'], row['D4'], row['D5'], row['D6'], row['D7'], row['D8']])
+
+            address = row['ID']
+            if address not in unique_addresses:
+                unique_addresses[address] = [0, 1]
+            else:
+                unique_addresses[address][1] += 1
+    for row in failed_rows:
+        address = row['ID']
+        d1 = int(row['D1'], 16)
+        d8 = int(row['D8'], 16)
+        if address not in unique_addresses:
+            unique_addresses[address] = [0, 1, 0, 0]  # Initialize with D1 and D8 matched counts set to 0
+        if is_d1_checksum:
+            unique_addresses[address][2] += 1
+        if is_d8_checksum:
+            unique_addresses[address][3] += 1
+
+    sorted_addresses = sorted(unique_addresses.items(), key=lambda x: x[1][0], reverse=True)
+    print("Address  Successes  Failures  Total  D1_Matched  D8_Matched")
+    for address, counts in sorted_addresses:
+        total = counts[0] + counts[1]
+        if total > 0:
+            print(f"{address}  {counts[0]}  {counts[1]}  {total}  {counts[2]}  {counts[3]}")

ChecksumTools/whatchecksum.py

@@ -0,0 +1,51 @@
+import argparse
+from typing import List
+import numpy as np
+
+def crc(data: List[int], len_: int, start_val: int, poly: int) -> int:
+    crc_val = start_val
+    for i in range(len_):
+        crc_val ^= data[i]
+        for _ in range(8):
+            if (crc_val & 0x80) != 0:
+                crc_val = ((crc_val << 1) ^ poly) & 0xFF
+            else:
+                crc_val <<= 1
+                crc_val &= 0xFF
+    return crc_val ^ 0xFF
+
+def is_sum_checksum_valid(messages: List[List[int]], correlation_threshold: float) -> bool:
+    for idx in [0, -1]:
+        sums = [sum(message[:idx] + message[idx+1:]) % 256 for message in messages]
+        checksums = [message[idx] for message in messages]
+        stddev_sums = np.std(sums)
+        stddev_checksums = np.std(checksums)
+        if stddev_sums == 0 or stddev_checksums == 0:
+            continue
+        correlation = np.corrcoef(sums, checksums)[0, 1]
+        if correlation > correlation_threshold:
+            return True
+    return False
+
+def detect_checksum_type(messages: List[List[int]], threshold: float = 0.5, correlation_threshold: float = 0.9):
+    total_messages = len(messages)
+    crc_matches = 0
+    if is_sum_checksum_valid(messages, correlation_threshold):
+        return "Sum"
+    for message in messages:
+        data_len = len(message) - 1
+        for poly in range(0x01, 0xFF):
+            for start in range(0x00, 0x100):
+                if crc(message[:data_len], data_len, start, poly) == message[-1]:
+                    crc_matches += 1
+    if crc_matches / total_messages > threshold:
+        return "CRC"
+    return "Unknown"
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Checksum type detector for CAN bus messages")
+    parser.add_argument("messages", nargs="+", help="CAN bus messages (AABBCCDDEEFF0011)")
+    args = parser.parse_args()
+    messages = [list(bytes.fromhex(msg)) for msg in args.messages]
+    checksum_type = detect_checksum_type(messages)
+    print(f"The detected checksum type is: {checksum_type}")