Python/cryptography/lfsr_bit_stream.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "65571230-ccb3-41a1-a361-31097b31bc5b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style>.container{width:100%}</style>\n"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%HTML\n",
    "<style>.container{width:100%}</style>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "d76a12a2-66f5-48b4-96c7-42cb8c8cae95",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[1, 0, 0, 0, 0, 1, 0, 1, 1, 0]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def id2bit(ls: list):\n",
    "    \"\"\"\n",
    "    Converts a list of indices to a binary representation (bit array).\n",
    "\n",
    "    Given a list of indices, this function creates a binary list where each index in\n",
    "    the input list is set to 1 in the output list, and all other positions are set to 0.\n",
    "    The output list is then reversed before returning.\n",
    "\n",
    "    Args:\n",
    "        ls (list): A list of indices where each index will be set to 1 in the output list.\n",
    "\n",
    "    Returns:\n",
    "        list: A list of binary values (0s and 1s), where each index in the input list corresponds\n",
    "              to a 1 in the output binary list, and all other indices are 0.\n",
    "    \"\"\"\n",
    "    if len(ls) == 0:\n",
    "        return [0, 0, 0, 0, 0, 0, 0, 0]  # Return a default 8-bit array\n",
    "    aa = [0 for i in range(max(ls) + 1)]\n",
    "    for i in ls:\n",
    "        aa[i] = 1\n",
    "    return aa[::-1]\n",
    "\n",
    "\n",
    "def bit2id(ls: list):\n",
    "    \"\"\"\n",
    "    Converts a binary list (bit array) to a list of indices where the value is 1.\n",
    "\n",
    "    This function iterates over the binary list and returns a list of indices where the binary value is 1.\n",
    "    The list is reversed before returning.\n",
    "\n",
    "    Args:\n",
    "        ls (list): A list of binary values (0s and 1s).\n",
    "\n",
    "    Returns:\n",
    "        list: A list of indices where the corresponding binary value in the input list is 1.\n",
    "    \"\"\"\n",
    "    ls = ls[::-1]\n",
    "    aa = []\n",
    "    for i in range(len(ls)):\n",
    "        if ls[i] == 1:\n",
    "            aa.append(i)\n",
    "    return aa[::-1]\n",
    "\n",
    "\n",
    "def XOR(*args):\n",
    "    \"\"\"\n",
    "    Performs bitwise XOR on a sequence of values.\n",
    "\n",
    "    This function takes any number of arguments and performs the XOR operation iteratively\n",
    "    across all the input values.\n",
    "\n",
    "    Args:\n",
    "        *args: A sequence of values (typically integers) on which the XOR operation will be applied.\n",
    "\n",
    "    Returns:\n",
    "        int: The result of applying the XOR operation across all input values.\n",
    "    \"\"\"\n",
    "    result = 0\n",
    "    for arg in args:\n",
    "        result ^= arg\n",
    "    return result\n",
    "\n",
    "\n",
    "class LFSR:\n",
    "    \"\"\"\n",
    "    A class representing a Linear Feedback Shift Register (LFSR).\n",
    "\n",
    "    This class models an LFSR, which generates a sequence of bits based on an initial state\n",
    "    and a feedback polynomial. The LFSR can be clocked to generate subsequent bits in the sequence.\n",
    "\n",
    "    Attributes:\n",
    "        seq (list): The current state (bit sequence) of the LFSR.\n",
    "        taps (list): The positions of the taps used for feedback calculation.\n",
    "\n",
    "    Methods:\n",
    "        clock(): Shifts the bits in the LFSR and computes the new bit based on the feedback.\n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self, start, poly):\n",
    "        \"\"\"\n",
    "        Initializes an LFSR with a start state and a feedback polynomial.\n",
    "\n",
    "        Args:\n",
    "            start (list): The initial state of the LFSR, represented as a list of bits (0s and 1s).\n",
    "            poly (list): A list representing the feedback polynomial, with 1s indicating the taps.\n",
    "\n",
    "        Raises:\n",
    "            ValueError: If the length of the start state does not match the polynomial length minus one.\n",
    "        \"\"\"\n",
    "        self.seq = start\n",
    "        self.taps = bit2id(poly[:-1])  # ignore the output tap (final bit)\n",
    "\n",
    "        if len(self.seq) != len(poly) - 1:\n",
    "            raise ValueError(\"Polynomial and start value length mismatch\")\n",
    "\n",
    "    def clock(self):\n",
    "        \"\"\"\n",
    "        Advances the LFSR by one clock cycle.\n",
    "\n",
    "        This method computes the feedback bit by XORing the bits at the tap positions,\n",
    "        shifts the state, and adds the feedback bit to the beginning of the sequence.\n",
    "        \"\"\"\n",
    "        feedback = XOR(*[self.seq[bit] for bit in self.taps])\n",
    "        self.seq = [feedback] + self.seq[:-1]\n",
    "\n",
    "\n",
    "class A51:\n",
    "    \"\"\"\n",
    "    A class representing the A5/1 stream cipher.\n",
    "\n",
    "    A51 is a stream cipher used in GSM encryption. It combines three LFSRs and uses a majority rule\n",
    "    to control which LFSRs are clocked. The output is the XOR of the last bits of the LFSRs.\n",
    "\n",
    "    Attributes:\n",
    "        lfsrs (list): A list of LFSR instances.\n",
    "        clock_bits (list): The bit positions used for clocking each LFSR.\n",
    "        lfsr_count (int): The number of LFSRs used in the cipher.\n",
    "\n",
    "    Methods:\n",
    "        majority(*bits): Computes the majority bit from a list of bits.\n",
    "        clock(): Advances the cipher and returns the next bit of the keystream.\n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self, lfsrs, clock_bits):\n",
    "        \"\"\"\n",
    "        Initializes the A51 cipher with a list of LFSRs and their clocking bits.\n",
    "\n",
    "        Args:\n",
    "            lfsrs (list): A list of LFSR instances used to generate the keystream.\n",
    "            clock_bits (list): A list indicating the bit positions in each LFSR to use for majority voting.\n",
    "        \"\"\"\n",
    "        self.lfsrs = lfsrs\n",
    "        self.clock_bits = clock_bits\n",
    "        self.lfsr_count = len(clock_bits)\n",
    "\n",
    "    def majority(self, *bits):\n",
    "        \"\"\"\n",
    "        Computes the majority bit from a sequence of bits.\n",
    "\n",
    "        This method determines the majority (1 or 0) from the given bits. If the number of 1s\n",
    "        is greater than or equal to half of the number of LFSRs, the majority bit is 1; otherwise, it is 0.\n",
    "\n",
    "        Args:\n",
    "            *bits: A sequence of bits (typically 0s and 1s) for which the majority is to be determined.\n",
    "\n",
    "        Returns:\n",
    "            int: The majority bit (0 or 1).\n",
    "        \"\"\"\n",
    "        ones = sum(i for i in bits if i == 1)\n",
    "        if ones >= self.lfsr_count / 2:\n",
    "            majority_bit = 1\n",
    "        else:\n",
    "            majority_bit = 0\n",
    "        return majority_bit\n",
    "\n",
    "    def clock(self):\n",
    "        \"\"\"\n",
    "        Advances the A51 cipher by one clock cycle and generates the next keystream bit.\n",
    "\n",
    "        This method computes the majority bit from the specified clocking positions of the LFSRs,\n",
    "        clocks the LFSRs if necessary, and outputs the XOR of the last bits of each LFSR as the next\n",
    "        bit of the keystream.\n",
    "\n",
    "        Returns:\n",
    "            int: The next bit in the keystream generated by the A51 cipher.\n",
    "        \"\"\"\n",
    "        majority = self.majority(\n",
    "            *[self.lfsrs[i].seq[self.clock_bits[i]] for i in range(self.lfsr_count)]\n",
    "        )\n",
    "        for i in range(self.lfsr_count):\n",
    "            if self.lfsrs[i].seq[self.clock_bits[i]] == majority:\n",
    "                self.lfsrs[i].clock()\n",
    "        out = XOR(*[int(i.seq[-1]) for i in self.lfsrs])\n",
    "        return out\n",
    "\n",
    "\n",
    "# Example usage\n",
    "lf1 = LFSR(start=[1, 0, 1, 1], poly=id2bit([4, 1]))\n",
    "lf2 = LFSR(start=[0, 1, 1, 1], poly=id2bit([4, 1]))\n",
    "lf3 = LFSR(start=[1, 0, 1, 0], poly=id2bit([4, 1]))\n",
    "a51 = A51(lfsrs=[lf1, lf2, lf3], clock_bits=[1, 2, 0])\n",
    "\n",
    "# Generate a keystream of 10 bits\n",
    "stream = [a51.clock() for i in range(10)]\n",
    "stream"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "182b2a83-d083-4296-a3bc-4d4f14dd8724",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "\n",
    "def write2txt_file(bitstream, filename):\n",
    "    \"\"\"\n",
    "    Writes a bitstream (string of '0's and '1's) to a text file.\n",
    "\n",
    "    This function opens a text file in append mode and writes the provided bitstream to it.\n",
    "\n",
    "    Args:\n",
    "        bitstream (str): A string of '0's and '1's representing the bitstream to be written.\n",
    "        filename (str): The path to the text file where the bitstream will be written.\n",
    "    \"\"\"\n",
    "    with open(filename, \"a\") as f:  # Open in append mode to continue writing\n",
    "        f.write(bitstream)\n",
    "\n",
    "\n",
    "def write2bin_file(bitstream, filename):\n",
    "    \"\"\"\n",
    "    Writes a bitstream (string of '0's and '1's) to a binary file.\n",
    "\n",
    "    This function converts the bitstream into bytes, pads it to ensure it's a multiple of 8 bits,\n",
    "    and then writes it to a binary file in append mode.\n",
    "\n",
    "    Args:\n",
    "        bitstream (str): A string of '0's and '1's representing the bitstream to be written.\n",
    "        filename (str): The path to the binary file where the bitstream will be written.\n",
    "    \"\"\"\n",
    "    byte_list = []\n",
    "\n",
    "    # Pad the bitstream if it's not a multiple of 8\n",
    "    padding = (8 - (len(bitstream) % 8)) % 8\n",
    "    bitstream += \"0\" * padding  # Add extra '0's to make the length a multiple of 8\n",
    "\n",
    "    for i in range(0, len(bitstream), 8):\n",
    "        byte = bitstream[i : i + 8]\n",
    "        byte_list.append(int(byte, 2))  # Convert 8 bits to an integer (byte)\n",
    "\n",
    "    # Append the bytes to the binary file\n",
    "    with open(filename, \"ab\") as f:  # 'ab' mode to append to the binary file\n",
    "        f.write(bytearray(byte_list))\n",
    "\n",
    "\n",
    "def gen_bit_stream(data: dict, target_size: int, file_path: str):\n",
    "    \"\"\"\n",
    "    Generates a keystream using the A51 cipher and writes it to a file.\n",
    "\n",
    "    This function initializes the LFSRs based on the provided data, generates a keystream\n",
    "    using the A51 cipher, and writes the generated bits to a text file or binary file\n",
    "    in chunks. It keeps track of the current size of the output file and prints progress\n",
    "    at each 10% interval.\n",
    "\n",
    "    Args:\n",
    "        data (dict): A dictionary containing information about the LFSRs, including their\n",
    "                     start values, polynomials, and clock positions.\n",
    "        target_size (int): The target size of the file in bytes. The function will stop once\n",
    "                            this size is reached.\n",
    "        file_path (str): The path to the output file where the generated bitstream will be written.\n",
    "    \"\"\"\n",
    "    # Initialize the LFSRs and A51 cipher\n",
    "    lfsrs = [LFSR(start=i[\"start\"], poly=i[\"poly\"]) for i in data]\n",
    "    a51 = A51(lfsrs=lfsrs, clock_bits=[i[\"clock\"] for i in data])\n",
    "\n",
    "    current_size = 0\n",
    "    bitstream_chunk = \"\"  # Chunk of bits to write periodically\n",
    "    chunk_size = (\n",
    "        10000  # Number of bits to generate at a time (can adjust for performance)\n",
    "    )\n",
    "    progress_interval = target_size // 10  # 1/10th of the target size (100 MB)\n",
    "    next_progress_checkpoint = progress_interval\n",
    "\n",
    "    # Generate bits until the target file size is reached\n",
    "    while current_size < target_size:\n",
    "        # Generate bits in chunks\n",
    "        for _ in range(chunk_size):\n",
    "            bitstream_chunk += str(a51.clock())\n",
    "\n",
    "        # Write the chunk to file\n",
    "        write2txt_file(bitstream_chunk, file_path)\n",
    "\n",
    "        # Clear the chunk and update the current file size\n",
    "        bitstream_chunk = \"\"\n",
    "        current_size = os.path.getsize(file_path)\n",
    "\n",
    "        # Check if the file size has crossed the 1/10th checkpoint\n",
    "        if current_size >= next_progress_checkpoint:\n",
    "            print(\n",
    "                f\"File size crossed {round(next_progress_checkpoint / (1024 * 1024), 2)} MB\"\n",
    "            )\n",
    "            next_progress_checkpoint += (\n",
    "                progress_interval  # Update to next 10% checkpoint\n",
    "            )\n",
    "\n",
    "    print(f\"File generation complete: {file_path} (target)\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "ebf2b473-4277-4b99-9935-96802dc52488",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "File size crossed 0.1 MB\n",
      "File generation complete: mine_gen_100MB.txt (target)\n"
     ]
    }
   ],
   "source": [
    "data = [\n",
    "    {\"start\": [0, 1, 0, 1, 1], \"poly\": id2bit([5, 2, 0]), \"clock\": 2},\n",
    "    {\"start\": [1, 0, 0, 1, 0], \"poly\": id2bit([5, 4, 3, 1, 0]), \"clock\": 3},\n",
    "    {\"start\": [0, 1, 1, 0, 0], \"poly\": id2bit([5, 4, 2, 1, 0]), \"clock\": 2},\n",
    "]\n",
    "gen_bit_stream(data, target_size=1 * 1024**2, file_path=\"mine_gen_100MB.txt\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "80acd699-63bc-4391-ac05-8166a843b1ca",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01038d4c-ab7f-486c-b300-a95187f4cd76",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}