PA193_RNG_2024_Tasks.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "ca8d45fc",
   "metadata": {},
   "source": [
    "# PA193 Seminar - RNGs\n",
    "This notebook contains code for several tasks treated in this seminar. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e1a8e708",
   "metadata": {},
   "source": [
    "# Imports: \n",
    " 1. Execute next cell."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "fe3c5279",
   "metadata": {},
   "outputs": [],
   "source": [
    "import time, math, secrets\n",
    "from collections import Counter\n",
    "from cryptography.hazmat.primitives import hashes\n",
    "from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d2e4de7d",
   "metadata": {},
   "source": [
    "# PRNG: state, seed, determinism \n",
    "\n",
    "We will work with PRNG implemented in [random](https://docs.python.org/3/library/random.html) package. See first 4 methods (`seed, setstate, getstate, randbytes`) in the documentation. \n",
    " 1. Import **random** package.\n",
    " 2. Generate (and print) 3 random bytes.  \n",
    " 3. Print out bytes in hexadecimal form (use `.hex()` method of bytes). Execute cell 2x - use Run or Ctrl+Enter.  \n",
    " 4. Now seed the generator with arbitrary value and execute 2x - you will see the same result (=RNG is deterministic and initilialized with seed).  \n",
    " 5. Save state of the generator (into `state` variable) of the PRNG -- insert line right after the seeding (before generation of bytes). Print out the state of the PRNG, it consists of multiple values.\n",
    " 6. Use the `state` to set up the PRNG and generate the same bytes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "ff090963",
   "metadata": {},
   "outputs": [],
   "source": [
    "# import the package then add more lines solving next lines"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "09da2a2d",
   "metadata": {},
   "source": [
    "# LCG: periodicity, seeding, brute force attack\n",
    "Standard PRNG functions are very fast but also very insecure. \n",
    " * In python, PRNG is [implemented](https://svn.python.org/projects/python/branches/release32-maint/Lib/random.py) in random module is [Mersenne Twister](https://en.wikipedia.org/wiki/Mersenne_Twister) with state formed by 625 32-bit integers. \n",
    " * In other languages (C, Java, Rust) LCG is typically used. Internal state of LCG is **single** value (state) updated iterativelly as $$state = (state*a+c) \\pmod m.$$ Overview of constants `a,c,m` used by the LCG for several languages can be found [LCG params and generators](https://en.wikipedia.org/wiki/Linear_congruential_generator).  \n",
    " 1. Implement LCG generator and instantiate `RNG` with constants $(a=3, c=1, m = 257)$ \n",
    " 2. Seed `RNG` with `0`, generate the sequence of bytes $S_1$ and find its period. Period should be 256 or 1 .\n",
    " 3. Find the missing number $s$ in $S_1.$ Reseed the generator with $s$, generate the sequence and find its period. The missing number is 128 - it produces constant sequence since $3*128+1 \\equiv 128 \\pmod {257}.$\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "bf827569",
   "metadata": {},
   "outputs": [],
   "source": [
    "class LCG:\n",
    "    def __init__(self, a, c, m):\n",
    "        self.a, self.c, self.m = a, c, m\n",
    "        # seed with some default seed here\n",
    "\n",
    "    def seed(self, seed):\n",
    "        self.state = seed % self.m\n",
    "\n",
    "    def rand(self) -> int:\n",
    "        # update the state\n",
    "        return 0 # replace with appropriate value \n",
    "    \n",
    "RNG = LCG(3, 1, 2**8+1)\n",
    "RNG.seed(int(time.time()))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fc49574f",
   "metadata": {},
   "source": [
    " 4. The plaintext `b'0123456789abcdef'` was encrypted by AES with key generated by LCG seeded by time. Find the key.\n",
    " * What is the maximal complexity of the attack? It suffices to generate whole cycle (256) and the test all 16B blocks, hence the complexity is 256 iterations. From practical point of view 256 + 16 bytes are generated and key candidates are sliced from that sequence of bytes. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "c4066886",
   "metadata": {},
   "outputs": [],
   "source": [
    "RNG.seed(int(time.time()))\n",
    "def LCG_bytes(num_bytes):\n",
    "    return bytes([RNG.rand()%256 for i in range(num_bytes)])\n",
    "\n",
    "K = LCG_bytes(16)\n",
    "AES_enc = Cipher(algorithms.AES(K), modes.ECB()).encryptor()\n",
    "plaintext = b'0123456789abcdef'\n",
    "ciphertext = AES_enc.update(plaintext)\n",
    "\n",
    "# find the key"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "843c453f",
   "metadata": {},
   "source": [
    "# LCG: Forward/backward predictability  \n",
    " 0. Parameters of LCG generators can be found here [LCG params and generators](https://en.wikipedia.org/wiki/Linear_congruential_generator) \n",
    " 1. Attacker knowns that the **glibc** generated number 1406932606. Why he is able to find the internal state of the **RNG**? What is the problem? \n",
    " 2. Use appropriate seed and generate next 9 values. \n",
    " 3. Are you able to create \"inverse\" LCG that goes in opposite directions? Start from the last value generated in 2. end up with the first one.  \n",
    "  - **HINT**: $x_{i+1} = a*x_{i}+c \\pmod m \\implies x_{i} = a^{-1}*x_{i+1}-(a^{-1}*c) \\pmod m$\n",
    "  - **HINT**: `pow(a, -1, m)` for python 3.6+ can be used to compute $a^{-1} \\pmod m$\n",
    "\n",
    " "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "f4152961",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "glibc = LCG(a=1103515245, c=12345, m=2**31)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6382731b",
   "metadata": {},
   "source": [
    "# Entropy: estimation\n",
    " 1. Use `time_entropy` and generate sequence `S` of 100 random blocks. It collects entropy from fluctuations of code execution time.\n",
    " 2. Compute histogram of generated values (use `Counter` from collections) and compute entropy of the source (`time_entropy`). Try different sizes of the sequence and you will see that the entropy estimation fluctuates. \n",
    "  * In case `time_entropy` generates same values uncomment the lines and try different values of `range`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "916a53a4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# x = []\n",
    "def time_entropy():\n",
    "    start = time.time_ns()\n",
    "#     for i in range(100):\n",
    "#         x.append(pow(random.choice([1,3,5]), -1, 2**31))\n",
    "    \n",
    "    delta = time.time_ns() - start\n",
    "    if delta < 256:\n",
    "        size = 1 \n",
    "    elif delta < 256**2:\n",
    "        size = 2\n",
    "    else:\n",
    "        size = 3\n",
    "    return delta.to_bytes(size, byteorder='little')\n",
    "        \n",
    "    \n",
    "def H_from_freqs(freqs):\n",
    "    return 0 "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a43171e0",
   "metadata": {},
   "source": [
    "# Entropy: values repetitions\n",
    " 1. How many random blocks with 32 bits of entropy (produced by system RNG) we need to generate to find one collision (repeated value/block). Use with `secret.token_bytes(??)` within `collision` function to generate random bytes. What should be used instead of `??`. Resulted number should be comparable to $2^{16}=65000$ (birthday paradox).  \n",
    " 2. Now use `generate_bytes` that uses biased src `time_entropy` instead of `secret.token_bytes(??)`. Function `time_entropy` produces approximately 2 bits of entropy per 1B. \n",
    " * Q: How many bytes we should generate using `generate_bytes` so they contain 32 bits of entropy? A: 16 bytes\n",
    " * Q: You can see that significantly less iterations (returned value of `collision`) are needed to obtain collision. What is the reason? A: the reason for that is that `time_entropy` generate values with different probabilities (while probabilities for `secret.token_bytes(??)` are equal)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "7c74dee8",
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_bytes(num_bytes):\n",
    "    buffer = bytes()\n",
    "    while len(buffer) < num_bytes:\n",
    "        buffer += time_entropy()\n",
    "    return buffer[:num_bytes]\n",
    "\n",
    "def collision(num_bytes):\n",
    "    S = []\n",
    "    for i in range(10**6): \n",
    "#         r = generate_bytes(num_bytes)       #uncomment this or next line\n",
    "#         r = secrets.token_bytes(num_bytes)\n",
    "        if r in S: \n",
    "            return len(S)\n",
    "        else:\n",
    "            S.append(r)\n",
    "            \n",
    "collision(0) # replace 0"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9b357582",
   "metadata": {},
   "source": [
    "# Entropy pool: processing pool\n",
    " 1. Use three different methods bytes slicing (`pool[a:b]`) `XOR`,`SHA1`,  to process 40 bytes of the pool to resulted random block of 20 bytes. Use `collision` function and  decide which will be better approach for producing different values.\n",
    " 2. When pool (in `add_event` method) becomes bigger then `maxpoolsize` it should be mininimized. But pool content should be replaced by block with the same entropy. Add the test of the pool size with appropriate processing and replacement to the `add_event` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "id": "44e5ffd3",
   "metadata": {},
   "outputs": [],
   "source": [
    "def SHA1(message: bytes):\n",
    "        digest = hashes.Hash(hashes.SHA1())\n",
    "        digest.update(message)\n",
    "        return digest.finalize() \n",
    "\n",
    "def XOR(bytes1, bytes2):\n",
    "    return bytes(a ^ b for (a, b) in zip(bytes1, bytes2))\n",
    "\n",
    "class EntropyPool(object):\n",
    "    def __init__(self, maxpoolsize=32) -> None:\n",
    "        self.maxsize = maxpoolsize\n",
    "        self.pool = bytes()\n",
    "    \n",
    "    def add_event(self, num_bytes) -> None:\n",
    "        self.pool += generate_bytes(num_bytes)\n",
    "        # test and pool processing\n",
    "        \n",
    "    def random(self):\n",
    "#         res = processed pool using SHA1, XOR, or slicing\n",
    "        self.pool = bytes()\n",
    "        return res"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "701d1e88",
   "metadata": {},
   "source": [
    "# CSPRNG: period, seeding, backdoor\n",
    " 1. Questions:\n",
    "     * What is the period of the `CRT_PRNG`? \n",
    "     * Can we replace AES by SHA1 and obtain same security?\n",
    "     * Internal state updated the same function (AES or SHA1) \n",
    " 2. Backdoored (designer knows the key of the generator) RNG generated 16B `nonce=b'z\\x94a\\x1e\\xe2\\x0e/\\r\\xe2\\x85\\xb6\\x94\\xca\\x1b\\xd1\\x91'` and then it was used to generate key `K` for AES(16B). Find the internal state of  `CRT_PRNG` and generated 16B key `K` (b'\\x97\\x14Y\\n~\\...')\n",
    " 3. Seed CRT_PRNG with appropriate amount of entropy from `EntropyPool`. \n",
    "  * Q: How many bytes should by in the pool to produce random key? A: The amount of entropy of the pool should be equal to size of the key. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "d94471a2",
   "metadata": {},
   "outputs": [],
   "source": [
    "class CRT_PRNG: \n",
    "    def __init__(self, seed):\n",
    "        self.seed(seed)\n",
    "        self.counter = 0\n",
    "        \n",
    "    def seed(self, seed):\n",
    "        cipher = Cipher(algorithms.AES(seed), modes.ECB())\n",
    "        self.AES = cipher.encryptor()\n",
    "        \n",
    "    def rand(self):\n",
    "        msg = self.counter.to_bytes(length=16, byteorder='little') \n",
    "        rnd_block = self.AES.update(msg)\n",
    "        self.counter += 1\n",
    "        return rnd_block\n",
    "    \n",
    "K = bytes(16) # zero key\n",
    "RNG_backdoored = CRT_PRNG(seed=K)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d44d981a",
   "metadata": {},
   "source": [
    "# Bonus: recent Minecraft RNG failure  \n",
    "The generator `JavaRNGMinecraft` implements simplified version of RNG used in Minecraft below. It directly outputs random integer `randomInteger`(no need to multiply floats back with `(1 << 24)`). The vulnerability and exploit is described in \n",
    "[Randar Explanation and Information](https://github.com/spawnmason/randar-explanation/blob/master/README.md).\n",
    "`public float nextFloat() {\n",
    "   this.seed = (this.seed * multiplier + addend) % modulus; // update the seed\n",
    "   int randomInteger = (int) (this.seed >> 24); // take the top 24 bits of the seed\n",
    "   return randomInteger / ((float) (1 << 24)); // divide it by 2^24 to get a number between 0 and 1\n",
    "}`\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "b4364bd6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3541470173938\n",
      "196853528143125\n",
      "170598886189052\n",
      "[211088, 11733384, 10168486]\n"
     ]
    }
   ],
   "source": [
    "class JavaRNGMinecraft:\n",
    "    def __init__(self, seed = 0):\n",
    "        self.seed(seed)\n",
    "\n",
    "    def seed(self, seed: int):\n",
    "        self.state = seed\n",
    "\n",
    "    def randomInteger(self) -> int:\n",
    "        self.state = (25214903917 * self.state + 11) % 2**48\n",
    "        print(self.state)\n",
    "        return (self.state >> 24)\n",
    "    \n",
    "RNG = JavaRNGMinecraft(int(time.time()))\n",
    "rnds = [RNG.randomInteger() for i in range(3)]\n",
    "print(rnds)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d9490cd4",
   "metadata": {},
   "source": [
    "1. Find the internal state of the generator. It can be found using different approaches slow brute force, fast LLL (see Randar), can you propose other method?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "f1b43505",
   "metadata": {},
   "outputs": [],
   "source": [
    "# find the state"
   ]
  }
 ],
 "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.10.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}