CSS-cipher/main.py

#!/usr/bin/env python3
from random import randint
from math import log2

class lfsr(object):
    def __init__(self, state, taps):
        self.state = state
        self.taps = taps

    # getters and setters (private attributes)
    def get_state(self):
        return self.__state
    def set_state(self, value):
        self.__state = value
    state = property(fget=get_state, fset=set_state, doc="lfsr state (current bits value in lfsr)")

    def get_taps(self):
        return self.__taps
    def set_taps(self, valeur):
        self.__taps = valeur
    taps = property(fget=get_taps, fset=set_taps, doc="lfsr taps (bit positions affecting next state)")

    def shift(self):
        '''
        calculate next state and return the output bit
        '''
        feedback = sum(self.state[tap] for tap in self.taps) % 2
        output = self.state[0]
        self.state = self.state[1:] + [feedback]
        return output


def test_lfsr17():
    print("test lfsr17")
    key = [randint(0, 1) for _ in range(16)] # first 16 bits
    key.append(1)  # prevent initial state from being {0}^17
    taps = [0, 14]
    lfsr17 = lfsr(key, taps)
    states = [lfsr17.state]
    for _ in range(2**17-2):
        lfsr17.shift()
        states.append(lfsr17.state)
    sorted_states = sorted(states, key=lambda x: tuple(x))
    for i in range(2**17-2):
        if sorted_states[i] == sorted_states[i+1]:  # compare each state with the next state in the sorted list, if 2 states are identical they should be next to each other
            print(f'state {sorted_states[i]} appears at least 2 times')
            return False
    n_state = len(sorted_states)
    n_state_log = log2(n_state+1)
    print(f'all {n_state} = 2^({n_state_log})-1 generated states are different')
    return True


def css_encrypt(text, key):
    taps17 = [0, 14]
    lfsr17 = lfsr((key[:16]+[1]), taps17)
    taps25 = [0, 3, 4, 12]
    lfsr25 = lfsr((key[16:]+[1]), taps25)
    cipher = 0
    carry = 0

    if isinstance(text, int):
        Bytes = text.to_bytes((text.bit_length() + 7) // 8, 'big')
    elif isinstance(text, bytes):
        Bytes = text
    else:
        raise TypeError("input text should be bytes or integer")

    for byte in Bytes:
        x_b2 = ""
        y_b2 = ""
        # Generate bytes from lfsr17 and lfsr25
        for _ in range(8):
            x_b2 += str(lfsr17.shift())
            y_b2 += str(lfsr25.shift())
        x = int(x_b2[::-1], 2)
        y = int(y_b2[::-1], 2)

        z = (x + y + carry) % 256
        carry = 1 if x + y > 255 else 0

        cipher_byte = z ^ byte
        # print(cipher_byte.to_bytes((cipher_byte.bit_length() + 7) // 8, 'big'))
        cipher = (cipher << 8) | cipher_byte
        cipher_bytes = b'\x00'*(len(Bytes) - len(cipher.to_bytes((cipher.bit_length() + 7) // 8, 'big'))) +cipher.to_bytes((cipher.bit_length() + 7) // 8, 'big') # padding
    return cipher_bytes

def test_encrypt():
    print("test encryption: text 0xffffffffff, key 0x0 ∈ {0, 1}^40")
    cipher = int.from_bytes(css_encrypt(0xffffffffff, [0]*40), byteorder='big')
    print(f'cipher: {hex(cipher)}')
    clear = int.from_bytes(css_encrypt(cipher, [0]*40))
    print(f'decrypted: {hex(clear)}')
    print(f'original text and decrypted message are the same: {clear==0xffffffffff}')

def gen_6_bytes(key=[randint(0, 1) for _ in range(40)]):
    text = b'\x00\x00\x00\x00\x00\x00'
    cipher = css_encrypt(text, key)
    return cipher

import multiprocessing

def attack_worker(start, end, Bytes, result_queue, stop_event):
    taps17 = [0, 14]
    taps25 = [0, 3, 4, 12]
    for i in range(start, end):
        if stop_event.is_set():
            return
        lfsr17_init = [int(bit) for bit in bin(i)[2:].zfill(16)]+[1]
        lfsr17 = lfsr(lfsr17_init, taps17)
        x = []
        for _ in range(3):
            x_bin = ""
            for _ in range(8):
                x_bin += str(lfsr17.shift())
            x.append(int(x_bin[::-1], 2))
        y = [(Bytes[0]-x[0])%256]
        c=1 if x[0]+y[0]>255 else 0
        y.append((Bytes[1]-(x[1]+c))%256)
        c=1 if x[1]+y[1]>255 else 0
        y.append((Bytes[2]-(x[2]+c))%256)
        lfsr25_init = [int(bit) for bit in (bin(y[0])[2:].zfill(8)[::-1] + bin(y[1])[2:].zfill(8)[::-1] + bin(y[2])[2:].zfill(8)[::-1] ) ]+[1]
        lfsr25 = lfsr(lfsr25_init, taps25)
        for _ in range(24):
            lfsr25.shift()
        for _ in range(3):
            x_bin = ""
            y_bin = ""
            for _ in range(8):
                x_bin += str(lfsr17.shift())
                y_bin += str(lfsr25.shift())
            x.append(int(x_bin[::-1], 2))
            y.append(int(y_bin[::-1], 2))
        c=1 if x[2]+y[2]>255 else 0
        z4 = (x[3]+y[3]+c)%256
        c=1 if x[3]+y[3]>255 else 0
        z5 = (x[4]+y[4]+c)%256
        c=1 if x[4]+y[4]>255 else 0
        z6 = (x[5]+y[5]+c)%256
        if z4 == Bytes[3] and z5 == Bytes[4] and z6 == Bytes[5]:
            key = bin(x[0])[2:].zfill(8)[::-1] + bin(x[1])[2:].zfill(8)[::-1] + bin(y[0])[2:].zfill(8)[::-1] + bin(y[1])[2:].zfill(8)[::-1] + bin(y[2])[2:].zfill(8)[::-1]
            result_queue.put(key)
            stop_event.set()
            return

def attack(Bytes=gen_6_bytes()):
    result_queue = multiprocessing.Queue()
    stop_event = multiprocessing.Event()
    processes = []
    num_cores = multiprocessing.cpu_count()
    max_upper_limit = 2**16
    chunk_size = max_upper_limit // num_cores

    for i in range(num_cores-1):
        start = i * chunk_size
        end = start + chunk_size
        process = multiprocessing.Process(target=attack_worker, args=(start, end, Bytes, result_queue, stop_event))
        processes.append(process)
        process.start()
    # last process
    start = (num_cores-1) * chunk_size
    end = max_upper_limit
    process = multiprocessing.Process(target=attack_worker, args=(start, end, Bytes, result_queue, stop_event))
    processes.append(process)
    process.start()

    for process in processes:
        process.join()

    while not result_queue.empty():
        key = result_queue.get()
        print(f'key found: \t{key}')
        return [int(bit) for bit in key]

def test_attack(n=1):
    success = 0
    print(f'testing attack in 2^16 against CSS {n} times (keys randomly generated each time)\n')
    for _ in range(n):
        key = [randint(0, 1) for _ in range(40)]
        # key = [1]*40
        key_string = ''.join(str(bit) for bit in key)
        print(f'key generated: \t{key_string}')
        Bytes = gen_6_bytes(key)
        found_key = attack(Bytes)
        if found_key == key:
            success += 1
        print()
    print(f'{success}/{n} success')


test_lfsr17()
print()
test_encrypt()
print()
test_attack(5)