Quantum File Management.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Install libaries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "liboqs not found, installing it in C:\\Users\\david\\_oqs\n",
      "5 4 3 2 1 \n",
      "Done installing liboqs\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "import random\n",
    "import platform\n",
    "import subprocess\n",
    "import sys\n",
    "import base64\n",
    "import hashlib\n",
    "import oqs\n",
    "import numpy as np\n",
    "from qiskit import QuantumCircuit, transpile, assemble\n",
    "from qiskit.visualization import plot_histogram\n",
    "from qiskit_aer import AerSimulator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "File Creation (50MB file with random data) and Comparison Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_random_file(filename, size_in_mb=50):\n",
    "    size_bytes = size_in_mb * 1024 * 1024\n",
    "    with open(filename, 'wb') as f:\n",
    "        f.write(os.urandom(size_bytes))\n",
    "    print(f\"{filename} created with size {size_in_mb} MB.\")\n",
    "\n",
    "def file_hash(filename, hash_algorithm='sha256'):\n",
    "    hash_func = hashlib.new(hash_algorithm)\n",
    "    with open(filename, \"rb\") as f:\n",
    "        while chunk := f.read(8192):\n",
    "            hash_func.update(chunk)\n",
    "    return hash_func.hexdigest()\n",
    "\n",
    "def compare_files(file1, file2):\n",
    "    f1_size = os.path.getsize(file1)\n",
    "    f1_hash = file_hash(file1)\n",
    "    f2_size = os.path.getsize(file2)\n",
    "    f2_hash = file_hash(file2)\n",
    "    \n",
    "    print(f\"File 1: Size = {f1_size} bytes, Hash = {f1_hash}\")\n",
    "    print(f\"File 2: Size = {f2_size} bytes, Hash = {f2_hash}\")\n",
    "    \n",
    "    return f1_size, f1_hash, f2_size, f2_hash\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Quantum Compression & Error Correction Function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to perform quantum compression and error correction using Qiskit\n",
    "def quantum_compression_error_correction(input_file, compressed_file, error_corrected_file):\n",
    "    # Create a simple quantum circuit\n",
    "    circuit = QuantumCircuit(8, 8)\n",
    "    circuit.h(range(8)) # Apply Hadamard gates\n",
    "    circuit.barrier()\n",
    "    circuit.measure(range(8), range(8))\n",
    "\n",
    "    # Execute the circuit\n",
    "    simulator = AerSimulator()\n",
    "    backend = simulator.run(\"qasm_simulator\")\n",
    "    job = transpile(circuit, backend) \n",
    "    result = job.result()\n",
    "\n",
    "    # Get the counts\n",
    "    counts = result.get_counts()\n",
    "    print(\"Quantum Circuit Results:\", counts)\n",
    "\n",
    "    # Save compressed data as an example (demonstration only)\n",
    "    with open(input_file, \"rb\") as f:\n",
    "        data = f.read()\n",
    "    \n",
    "    # Simulate compression (for demonstation purposes, not real compression)\n",
    "    with open(compressed_file, \"wb\") as f:\n",
    "        f.write(data[:len(data)//2])\n",
    "    \n",
    "    # Simulate error correction (for demonstraton purposes, not real error correction)\n",
    "    with open(error_corrected_file, \"wb\") as f:\n",
    "        f.write(data[:len(data)//2])\n",
    "    \n",
    "    print(\"Quantum compression and error correction completed\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Quantum Encryption & Decryption Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to handle post-quantum encryption and secure key storage\n",
    "def pqc_kyber(input_file):\n",
    "    # Initialize Kyber KEM algorithm\n",
    "    kem = oqs.KeyEncapsulation(\"Kyber1024\")\n",
    "\n",
    "    # Generate keypair\n",
    "    public_key = kem.generate_keypair()\n",
    "\n",
    "    # Encapsulate the secret\n",
    "    ciphertext, shared_secret_enc = kem.encap_secret(public_key)\n",
    "    print(\"Shared Secret (Encapsulator):\", shared_secret_enc.hex())\n",
    "\n",
    "    # Store the shared secret using TPM or DPAPI depending on the OS\n",
    "    if platform.system() == \"Linux\":\n",
    "        \n",
    "        # Store the key with tpm2-tools\n",
    "        print(\"Linux detected: Using TPM2 tools\")\n",
    "        with open(\"shared_secret.bin\", \"wb\") as f:\n",
    "            f.write(shared_secret_enc)\n",
    "        \n",
    "        # Store the key using TPM\n",
    "        os.system(\"tpm2_nvdefine 0x1500016 -s 64 -a 'ownerread|ownerwrite'\")\n",
    "        os.system(\"tpm2_nvwrite 0x1500016 -i shared_secret.bin\")\n",
    "        print(\"Shared secret securely stored using TPM 2.0\")\n",
    "\n",
    "    elif platform.system() == \"Windows\":\n",
    "        print(\"Windows detected: Using DPAPI\")\n",
    "        import ctypes\n",
    "        from ctypes import wintypes\n",
    "\n",
    "        # Use Windows Data Protection API (DPAPI) to securely store the key\n",
    "        CRYPTPROTECT_UI_FORBIDDEN = 0x01\n",
    "        crypt32 = ctypes.windll.crypt32\n",
    "        kernel32 = ctypes.windll.kernel32\n",
    "\n",
    "        class DATA_BLOB(ctypes.Structure):\n",
    "            _fields_ = [(\"cbData\", wintypes.DWORD),\n",
    "                        (\"pbData\", ctypes.POINTER(ctypes.c_byte))]\n",
    "        \n",
    "        def protect_data(data: bytes) -> bytes:\n",
    "            blob_in = DATA_BLOB(len(data), ctypes.cast(ctypes.create_string_buffer(data), ctypes.POINTER(ctypes.c_byte)))\n",
    "            blob_out = DATA_BLOB()\n",
    "\n",
    "            result = crypt32.CryptProtectData(ctypes.byref(blob_in), None, None, None, None, CRYPTPROTECT_UI_FORBIDDEN, ctypes.byref(blob_out))\n",
    "            if not result:\n",
    "                raise ctypes.WinError()\n",
    "            \n",
    "            encrypted_data = ctypes.string_at(blob_out.pbData, blob_out.cbData)\n",
    "            kernel32.LocalFree(blob_out.pbData)\n",
    "            return encrypted_data\n",
    "    \n",
    "        encrypted_shared_secret = protect_data(shared_secret_enc)\n",
    "        with open(\"shared_secret_encrypted.bin\", \"wb\") as f:\n",
    "            f.write(encrypted_shared_secret)\n",
    "        print(\"Shared secret securely stored using DPAPI\")\n",
    "    \n",
    "    # Simulate decapsulation and key retrieval\n",
    "    shared_secret_dec = kem.decap_secret(ciphertext)\n",
    "    print(\"Shared Secret (Decapsulator):\", shared_secret_dec.hex())\n",
    "\n",
    "    # Compare shared secrets to verify correctness\n",
    "    assert shared_secret_enc == shared_secret_dec\n",
    "    print(\"Quantum-safe encryption and decryption using Kyber with TPM/DPAPI was successfull!!\")\n",
    "\n",
    "    return shared_secret_enc, encrypted_shared_secret\n",
    "\n",
    "# Function to decrypt the encrypted file using Kyber\n",
    "def decrypt_file(encrypted_file, decrypted_file, shared_secret_enc):\n",
    "    kem = oqs.KeyEncapsulation(\"Kyber1024\")\n",
    "\n",
    "    # Decapsulate the shared secret\n",
    "    shared_secret_dec = kem.decap_secret(shared_secret_enc)\n",
    "    print(\"Shared Secret (Decapsulator):\", shared_secret_dec.hex())\n",
    "\n",
    "    # Simulate the decryption (actual decryption should be implemented)\n",
    "    with open(encrypted_file, \"rb\") as f:\n",
    "        encrypted_data = f.read()\n",
    "    \n",
    "    # Simulate decryption (for demonstration purposes, not real decryption)\n",
    "    with open(decrypt_file, \"wb\") as f:\n",
    "        f.write(encrypted_data)\n",
    "    \n",
    "    print(f\"File decrypted and saved as {decrypted_file}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Quantum Decompression and Classical Error Correction Function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to decompress the file and perform classical error correction\n",
    "def decompress_and_correct(input_file, decompressed_file):\n",
    "    # Read the compressed file\n",
    "    with open(input_file, \"rb\") as f:\n",
    "        compressed_data = f.read()\n",
    "    \n",
    "    # Simulate decompression (for demonstration purposes, not real decompression)\n",
    "    decompressed_data = compressed_data * 2  # Placeholder for real decompression\n",
    "    \n",
    "    # Simulate classical error correction (for demonstration purposes, not real correction)\n",
    "    error_corrected_data = decompressed_data[:len(decompressed_data) // 2]\n",
    "    \n",
    "    # Save the decompressed and corrected data\n",
    "    with open(decompressed_file, \"wb\") as f:\n",
    "        f.write(error_corrected_data)\n",
    "    \n",
    "    print(f\"File decompressed and error-corrected, saved as {decompressed_file}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Main script for calling functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "random_file.bin created with size 50 MB.\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "C:\\Users\\david\\AppData\\Local\\Temp\\ipykernel_9728\\2285568480.py:12: DeprecationWarning: The `transpile` function will stop supporting inputs of type `BackendV1` ( <qiskit_aer.jobs.aerjob.AerJob object at 0x0000022D2BD988C0> ) in the `backend` parameter in a future release no earlier than 2.0. `BackendV1` is deprecated and implementations should move to `BackendV2`.\n",
      "  job = transpile(circuit, backend)\n"
     ]
    },
    {
     "ename": "AttributeError",
     "evalue": "'AerJob' object has no attribute 'configuration'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mAttributeError\u001b[0m                            Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[6], line 12\u001b[0m\n\u001b[0;32m      9\u001b[0m create_random_file(original_filename)\n\u001b[0;32m     11\u001b[0m \u001b[38;5;66;03m# Step 2: Perform quantum compression and error correction\u001b[39;00m\n\u001b[1;32m---> 12\u001b[0m \u001b[43mquantum_compression_error_correction\u001b[49m\u001b[43m(\u001b[49m\u001b[43moriginal_filename\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mcompressed_filename\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43merror_corrected_filename\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m     14\u001b[0m \u001b[38;5;66;03m# Step 3: Perform post-quantum encryption and secure key storage\u001b[39;00m\n\u001b[0;32m     15\u001b[0m shared_secret_enc, encrypted_shared_secret \u001b[38;5;241m=\u001b[39m pqc_kyber(error_corrected_filename)\n",
      "Cell \u001b[1;32mIn[3], line 12\u001b[0m, in \u001b[0;36mquantum_compression_error_correction\u001b[1;34m(input_file, compressed_file, error_corrected_file)\u001b[0m\n\u001b[0;32m     10\u001b[0m simulator \u001b[38;5;241m=\u001b[39m AerSimulator()\n\u001b[0;32m     11\u001b[0m backend \u001b[38;5;241m=\u001b[39m simulator\u001b[38;5;241m.\u001b[39mrun(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mqasm_simulator\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[1;32m---> 12\u001b[0m job \u001b[38;5;241m=\u001b[39m \u001b[43mtranspile\u001b[49m\u001b[43m(\u001b[49m\u001b[43mcircuit\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mbackend\u001b[49m\u001b[43m)\u001b[49m \n\u001b[0;32m     13\u001b[0m result \u001b[38;5;241m=\u001b[39m job\u001b[38;5;241m.\u001b[39mresult()\n\u001b[0;32m     15\u001b[0m \u001b[38;5;66;03m# Get the counts\u001b[39;00m\n",
      "File \u001b[1;32md:\\Projects\\Python\\Post-Quantum File Management\\.venv\\Lib\\site-packages\\qiskit\\compiler\\transpiler.py:339\u001b[0m, in \u001b[0;36mtranspile\u001b[1;34m(circuits, backend, basis_gates, inst_map, coupling_map, backend_properties, initial_layout, layout_method, routing_method, translation_method, scheduling_method, instruction_durations, dt, approximation_degree, timing_constraints, seed_transpiler, optimization_level, callback, output_name, unitary_synthesis_method, unitary_synthesis_plugin_config, target, hls_config, init_method, optimization_method, ignore_backend_supplied_default_methods, num_processes)\u001b[0m\n\u001b[0;32m    326\u001b[0m     \u001b[38;5;28;01mwith\u001b[39;00m warnings\u001b[38;5;241m.\u001b[39mcatch_warnings():\n\u001b[0;32m    327\u001b[0m         \u001b[38;5;66;03m# This is a temporary conversion step to allow for a smoother transition\u001b[39;00m\n\u001b[0;32m    328\u001b[0m         \u001b[38;5;66;03m# to a fully target-based transpiler pipeline while maintaining the behavior\u001b[39;00m\n\u001b[1;32m   (...)\u001b[0m\n\u001b[0;32m    331\u001b[0m         \u001b[38;5;66;03m#   removed once it gets removed:\u001b[39;00m\n\u001b[0;32m    332\u001b[0m         \u001b[38;5;66;03m#   https://github.com/Qiskit/qiskit/pull/12850\u001b[39;00m\n\u001b[0;32m    333\u001b[0m         warnings\u001b[38;5;241m.\u001b[39mfilterwarnings(\n\u001b[0;32m    334\u001b[0m             \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mignore\u001b[39m\u001b[38;5;124m\"\u001b[39m,\n\u001b[0;32m    335\u001b[0m             category\u001b[38;5;241m=\u001b[39m\u001b[38;5;167;01mDeprecationWarning\u001b[39;00m,\n\u001b[0;32m    336\u001b[0m             message\u001b[38;5;241m=\u001b[39m\u001b[38;5;124mr\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m.+qiskit\u001b[39m\u001b[38;5;124m\\\u001b[39m\u001b[38;5;124m.providers\u001b[39m\u001b[38;5;124m\\\u001b[39m\u001b[38;5;124m.backend_compat\u001b[39m\u001b[38;5;124m\\\u001b[39m\u001b[38;5;124m.BackendV2Converter.+\u001b[39m\u001b[38;5;124m\"\u001b[39m,\n\u001b[0;32m    337\u001b[0m             module\u001b[38;5;241m=\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mqiskit\u001b[39m\u001b[38;5;124m\"\u001b[39m,\n\u001b[0;32m    338\u001b[0m         )\n\u001b[1;32m--> 339\u001b[0m         backend \u001b[38;5;241m=\u001b[39m \u001b[43mBackendV2Converter\u001b[49m\u001b[43m(\u001b[49m\u001b[43mbackend\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m    341\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m (\n\u001b[0;32m    342\u001b[0m     scheduling_method \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m\n\u001b[0;32m    343\u001b[0m     \u001b[38;5;129;01mand\u001b[39;00m backend \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m\n\u001b[0;32m    344\u001b[0m     \u001b[38;5;129;01mand\u001b[39;00m target \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m\n\u001b[0;32m    345\u001b[0m     \u001b[38;5;129;01mand\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m instruction_durations\n\u001b[0;32m    346\u001b[0m ):\n\u001b[0;32m    347\u001b[0m     warnings\u001b[38;5;241m.\u001b[39mwarn(\n\u001b[0;32m    348\u001b[0m         \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mWhen scheduling circuits without backend,\u001b[39m\u001b[38;5;124m\"\u001b[39m\n\u001b[0;32m    349\u001b[0m         \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m \u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124minstruction_durations\u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124m should be usually provided.\u001b[39m\u001b[38;5;124m\"\u001b[39m,\n\u001b[0;32m    350\u001b[0m         \u001b[38;5;167;01mUserWarning\u001b[39;00m,\n\u001b[0;32m    351\u001b[0m     )\n",
      "File \u001b[1;32md:\\Projects\\Python\\Post-Quantum File Management\\.venv\\Lib\\site-packages\\qiskit\\providers\\backend_compat.py:381\u001b[0m, in \u001b[0;36mBackendV2Converter.__init__\u001b[1;34m(self, backend, name_mapping, add_delay, filter_faulty)\u001b[0m\n\u001b[0;32m    362\u001b[0m \u001b[38;5;250m\u001b[39m\u001b[38;5;124;03m\"\"\"Initialize a BackendV2 converter instance based on a BackendV1 instance.\u001b[39;00m\n\u001b[0;32m    363\u001b[0m \n\u001b[0;32m    364\u001b[0m \u001b[38;5;124;03mArgs:\u001b[39;00m\n\u001b[1;32m   (...)\u001b[0m\n\u001b[0;32m    378\u001b[0m \u001b[38;5;124;03m        those from the output target.\u001b[39;00m\n\u001b[0;32m    379\u001b[0m \u001b[38;5;124;03m\"\"\"\u001b[39;00m\n\u001b[0;32m    380\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_backend \u001b[38;5;241m=\u001b[39m backend\n\u001b[1;32m--> 381\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_config \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_backend\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mconfiguration\u001b[49m()\n\u001b[0;32m    382\u001b[0m \u001b[38;5;28msuper\u001b[39m()\u001b[38;5;241m.\u001b[39m\u001b[38;5;21m__init__\u001b[39m(\n\u001b[0;32m    383\u001b[0m     provider\u001b[38;5;241m=\u001b[39mbackend\u001b[38;5;241m.\u001b[39mprovider,\n\u001b[0;32m    384\u001b[0m     name\u001b[38;5;241m=\u001b[39mbackend\u001b[38;5;241m.\u001b[39mname(),\n\u001b[1;32m   (...)\u001b[0m\n\u001b[0;32m    387\u001b[0m     backend_version\u001b[38;5;241m=\u001b[39m\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_config\u001b[38;5;241m.\u001b[39mbackend_version,\n\u001b[0;32m    388\u001b[0m )\n\u001b[0;32m    389\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_options \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_backend\u001b[38;5;241m.\u001b[39m_options\n",
      "\u001b[1;31mAttributeError\u001b[0m: 'AerJob' object has no attribute 'configuration'"
     ]
    }
   ],
   "source": [
    "if __name__ == \"__main__\":\n",
    "    # Step 1: Create 50MB file of random characters\n",
    "    original_filename = \"random_file.bin\"\n",
    "    compressed_filename = \"compressed_file.bin\"\n",
    "    error_corrected_filename = \"error_corrected_file.bin\"\n",
    "    decrypted_filename = \"decrypted_file.bin\"\n",
    "    final_filename = \"final_file.bin\"\n",
    "\n",
    "    create_random_file(original_filename)\n",
    "\n",
    "    # Step 2: Perform quantum compression and error correction\n",
    "    quantum_compression_error_correction(original_filename, compressed_filename, error_corrected_filename)\n",
    "\n",
    "    # Step 3: Perform post-quantum encryption and secure key storage\n",
    "    shared_secret_enc, encrypted_shared_secret = pqc_kyber(error_corrected_filename)\n",
    "\n",
    "    # Step 4: Decrypt the encrypted file\n",
    "    decrypt_file(error_corrected_filename, decrypted_filename, shared_secret_enc)\n",
    "\n",
    "    # Step 5: Decompress the file and perform classical error correction\n",
    "    decompress_and_correct(compressed_filename, final_filename)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}