import gradio as gr import subprocess from concrete.ml.deployment import FHEModelClient from requests import head import numpy import os from pathlib import Path import requests import json import base64 import shutil import time import pandas as pd import pickle import numpy as np import pdb # This repository's directory REPO_DIR = Path(__file__).parent subprocess.Popen(["uvicorn", "server:app"], cwd=REPO_DIR) # subprocess.Popen(["uvicorn", "server:app", "--port", "3000"], cwd=REPO_DIR) # if not exists, create a directory for the FHE keys called .fhe_keys if not os.path.exists(".fhe_keys"): os.mkdir(".fhe_keys") # if not exists, create a directory for the tmp files called tmp if not os.path.exists("tmp"): os.mkdir("tmp") # Wait 5 sec for the server to start time.sleep(5) # Encrypted data limit for the browser to display # (encrypted data is too large to display in the browser) ENCRYPTED_DATA_BROWSER_LIMIT = 500 N_USER_KEY_STORED = 20 #Evaluation Key eval_key = [] #Encodings vector encodings = [] #User ID user_id = [] def clean_tmp_directory(): # Allow 20 user keys to be stored. # Once that limitation is reached, deleted the oldest. path_sub_directories = sorted( [f for f in Path(".fhe_keys/").iterdir() if f.is_dir()], key=os.path.getmtime ) user_ids = [] if len(path_sub_directories) > N_USER_KEY_STORED: n_files_to_delete = len(path_sub_directories) - N_USER_KEY_STORED for p in path_sub_directories[:n_files_to_delete]: user_ids.append(p.name) shutil.rmtree(p) list_files_tmp = Path("tmp/").iterdir() # Delete all files related to user_id for file in list_files_tmp: for user_id in user_ids: if file.name.endswith(f"{user_id}.npy"): file.unlink() def keygen(eval_key, user_id): # Clean tmp directory if needed clean_tmp_directory() print("Initializing FHEModelClient...") # Let's create a user_id user_id = numpy.random.randint(0, 2**32) fhe_api = FHEModelClient(f"fhe_model", f".fhe_keys/{user_id}") fhe_api.load() # Generate a fresh key fhe_api.generate_private_and_evaluation_keys(force=True) evaluation_key = fhe_api.get_serialized_evaluation_keys() numpy.save(f"tmp/tmp_evaluation_key_{user_id}.npy", evaluation_key) eval_key = list(evaluation_key)[:ENCRYPTED_DATA_BROWSER_LIMIT] return eval_key, user_id def encode_quantize(test_file, eval_key, encodings): ugly = ['Machine', 'SizeOfOptionalHeader', 'Characteristics', 'MajorLinkerVersion', 'MinorLinkerVersion', 'SizeOfCode', 'SizeOfInitializedData', 'SizeOfUninitializedData', 'AddressOfEntryPoint', 'BaseOfCode', 'BaseOfData', 'ImageBase', 'SectionAlignment', 'FileAlignment', 'MajorOperatingSystemVersion', 'MinorOperatingSystemVersion', 'MajorImageVersion', 'MinorImageVersion', 'MajorSubsystemVersion', 'MinorSubsystemVersion', 'SizeOfImage', 'SizeOfHeaders', 'CheckSum', 'Subsystem', 'DllCharacteristics', 'SizeOfStackReserve', 'SizeOfStackCommit', 'SizeOfHeapReserve', 'SizeOfHeapCommit', 'LoaderFlags', 'NumberOfRvaAndSizes', 'SectionsNb', 'SectionsMeanEntropy', 'SectionsMinEntropy', 'SectionsMaxEntropy', 'SectionsMeanRawsize', 'SectionsMinRawsize', 'SectionsMeanVirtualsize', 'SectionsMinVirtualsize', 'SectionMaxVirtualsize', 'ImportsNbDLL', 'ImportsNb', 'ImportsNbOrdinal', 'ExportNb', 'ResourcesNb', 'ResourcesMeanEntropy', 'ResourcesMinEntropy', 'ResourcesMaxEntropy', 'ResourcesMeanSize', 'ResourcesMinSize', 'ResourcesMaxSize', 'LoadConfigurationSize', 'VersionInformationSize'] fhe_api = FHEModelClient(f"fhe_model", f".fhe_keys/{eval_key}") fhe_api.load() from PE_main import extract_infos # expect [1, 53] but we get (53) # pdb.set_trace() # features = pickle.loads(open(os.path.join("features.pkl"), "rb").read()) encodings = extract_infos(test_file) encodings = list(map(lambda x: encodings[x], ugly)) encodings = np.array(encodings).reshape(1, -1) return encodings def encrypt_encoded_quantize(encodings, user_id, eval_key): fhe_api = FHEModelClient(f"fhe_model", f".fhe_keys/{user_id}") fhe_api.load() encodings = np.array(encodings) print(encodings) quantized_encodings = fhe_api.model.quantize_input(encodings).astype(numpy.uint8) print(quantized_encodings) encrypted_quantized_encoding = fhe_api.quantize_encrypt_serialize(encodings) print(encrypted_quantized_encoding) numpy.save( f"tmp/tmp_encrypted_quantized_encoding_{user_id}.npy", encrypted_quantized_encoding, ) # Compute size encrypted_quantized_encoding_shorten = list(encrypted_quantized_encoding)[:ENCRYPTED_DATA_BROWSER_LIMIT] encrypted_quantized_encoding_shorten_hex = "".join(f"{i:02x}" for i in encrypted_quantized_encoding_shorten) return (encrypted_quantized_encoding_shorten_hex) def run_fhe(user_id): encoded_data_path = Path(f"tmp/tmp_encrypted_quantized_encoding_{user_id}.npy") encrypted_quantized_encoding = numpy.load(encoded_data_path) # Read evaluation_key from the file evaluation_key = numpy.load(f"tmp/tmp_evaluation_key_{user_id}.npy") # Use base64 to encode the encodings and evaluation key encrypted_quantized_encoding = base64.b64encode(encrypted_quantized_encoding).decode() encoded_evaluation_key = base64.b64encode(evaluation_key).decode() query = {} query["evaluation_key"] = encoded_evaluation_key query["encrypted_encoding"] = encrypted_quantized_encoding headers = {"Content-type": "application/json"} response = requests.post( "http://localhost:8000/predict", data=json.dumps(query), headers=headers, ) encrypted_prediction = base64.b64decode(response.json()["encrypted_prediction"]) numpy.save(f"tmp/tmp_encrypted_prediction_{user_id}.npy", encrypted_prediction) encrypted_prediction_shorten = list(encrypted_prediction)[:ENCRYPTED_DATA_BROWSER_LIMIT] encrypted_prediction_shorten_hex = "".join(f"{i:02x}" for i in encrypted_prediction_shorten) return encrypted_prediction_shorten_hex def decrypt_prediction(user_id): encoded_data_path = Path(f"tmp/tmp_encrypted_prediction_{user_id}.npy") # Read encrypted_prediction from the file encrypted_prediction = numpy.load(encoded_data_path).tobytes() fhe_api = FHEModelClient(f"fhe_model", f".fhe_keys/{user_id}") fhe_api.load() # We need to retrieve the private key that matches the client specs (see issue #18) fhe_api.generate_private_and_evaluation_keys(force=False) predictions = fhe_api.deserialize_decrypt_dequantize(encrypted_prediction) if(predictions[0][0] >= 0.5): return "Safe file", predictions else: return "Malware", predictions if __name__ == "__main__": """ with gr.Blocks() as demo: print("Starting the FHE Model") inputs = [gr.File(label="Test File")] outputs = [ gr.Textbox(label="Evaluation Key"), gr.Textbox(label="Encodings"), gr.Textbox(label="Encrypted Quantized Encoding"), gr.Textbox(label="Encrypted Prediction"), ] run_bottn = gr.Button(label="Run") run_bottn.click( fn=process_pipeline, # Pass process_pipeline directly inputs=inputs, outputs=outputs ) demo.launch() """ print("🚀 Starting the ClairVault demo...") with gr.Blocks(css=".gradio-container { font-size: 20px; }") as demo: gr.Markdown( """
GitHub — Documentation — Community — @ClairVault
""" ) gr.Markdown("## 📋 Executive Summary") gr.Markdown( """ ### ClairVault is a **malware scanning service**, cloud-based and machine-learning enabled, that leverages fully homomorphic encryption (FHE) to securely scan files without seeing your data. """ ) gr.Markdown("## 🚨 The Problem") gr.Markdown( """ 1. Privacy: do you really trust the Russian-based Kaspersky and that there is no back-door built in McAfee by Oncle Sam? - Require access to plaintext data, posing privacy and security risks - Often closed-source, lacking transparency 2. Protection: - The malware scanner itself might be compromised - Require frequent local updates - Mostly rule-based → need machine learning - but mostly on cloud 3. Resources - Consume local resources (Windows Defender only runs ML based models on the cloud) """ ) gr.Markdown("## 💡 Our Solution: ClairVault") gr.Markdown( """ Key features include: - Local extraction of features and encryption of user data (files, logs) - Transmission of encrypted data to secure, open-source cloud servers - Malware classification performed on encrypted data - Return of encrypted classification results """ ) gr.Markdown("### 🛠️ Technical Implementation") gr.Markdown( """ - **Encryption Method**: Fully Homomorphic Encryption using the TFHE (Fast Fully Homomorphic Encryption over the Torus) library - **Machine Learning Model**: Linear classifier optimized for FHE computations using ConcreteML - **Performance Metrics**: - Encrypted file scanning: ~30 seconds per MB - Plaintext file scanning: ~0.5 seconds per MB - Estimated time to scan 1GB: ~8.5 hours (encrypted) vs. ~8.5 minutes (plaintext) *Note: These are preliminary figures based on our proof-of-concept. We aim to significantly improve performance in future iterations.* """ ) gr.Markdown("# 🗝️ Step 1: Generate the keys") b_gen_key = gr.Button("🔑 Generate the keys and send public part to server") evaluation_key = gr.Textbox( label="Evaluation key (truncated):", max_lines=4, interactive=False, ) gr.Markdown("# 📤 Step 2: Upload a file for scanning") gr.Markdown("## Client side") gr.Markdown( "Upload a file you want to scan for malware. ClairVault will encrypt it locally before sending it to the cloud." ) file_input = gr.File(label="Upload a file:", file_count = "single", value="./smallexe64.exe") gr.Markdown("# 📥 Step 3: Extract executable file features") b_extract = gr.Button("📥 Extract features and save") extracted_vector = gr.JSON( label="Extracted vector:", ) gr.Markdown("# 🔒 Step 4: Encrypt the file with the private key") b_encrypt_file = gr.Button( "🔐 Encrypt the file and send to server" ) encrypted_file = gr.Textbox( label="Encrypted file content (truncated):", max_lines=4, interactive=False, ) gr.Markdown("# 🖥️ Step 5: Run the FHE-based malware scan") gr.Markdown("## Server side") gr.Markdown( "The encrypted file is received by the server. Using the evaluation key and FHE, the server can perform the malware scan directly on the encrypted data. Once the scan is finished, the server returns the encrypted result to the client." ) b_run_fhe_scan = gr.Button("🛡️ Run FHE-based malware scan") encrypted_scan_result = gr.Textbox( label="Encrypted scan result (truncated):", max_lines=4, interactive=False, ) gr.Markdown("# 🔓 Step 6: Decrypt the scan result") gr.Markdown("## Client side") gr.Markdown( "The encrypted scan result is sent back to the client, who can finally decrypt it with their private key. Only the client is aware of the original file content and the scan result." ) b_decrypt_result = gr.Button("🔍 Decrypt scan result") user_id_input = gr.Number(visible=False) scan_result = gr.Textbox(label="Scan Result:") eval_key_input = gr.Textbox(value=eval_key, visible=False) # Button for key generation b_gen_key.click(fn=keygen, inputs=[eval_key_input, user_id_input], outputs=[evaluation_key, user_id_input]) encodings_input = gr.Textbox(value=encodings, visible=False) # Button to extract vector b_extract.click( fn=encode_quantize, inputs=[file_input, eval_key_input, encodings_input], outputs=[extracted_vector], ) # Button to encrypt file b_encrypt_file.click( fn=encrypt_encoded_quantize, inputs=[extracted_vector, user_id_input, eval_key_input], outputs=[encrypted_file], ) # Button to run FHE-based malware scan b_run_fhe_scan.click(fn=run_fhe, inputs=[user_id_input], outputs=[encrypted_scan_result]) # Button to decrypt the scan result b_decrypt_result.click(fn=decrypt_prediction, inputs=[user_id_input], outputs=[scan_result]) gr.Markdown( "ClairVault is built using advanced Fully Homomorphic Encryption techniques to ensure your data remains private and secure throughout the entire malware scanning process." ) gr.Markdown("## 🌍 Market Opportunity") gr.Markdown( """ The global cybersecurity market is valued at $60 billion in 2024 with an annual growth rate of 15% projected by Morgan Stanley. In 2023, there were 6 billion cyberattacks, a 10% increase from 2022. **Target Industries**: Finance, Healthcare, Government, Legal Services, Individuals **Estimated TAM (Total Addressable Market) for Privacy-Preserving Malware Scanning**: While exact figures for malware scanning are not available, we estimate it to be approximately 10% of the total cybersecurity market, or $6 billion. (Kaspersky has 700 million of revenue, Norton 1.5 billion, McAfee 2.5 billion.) Based on the critical need in our target industries we estimate privacy-preserving to be 5-10% of that, approximately $300-600 million. """ ) gr.Markdown("## 🏆 Competitive Advantage") gr.Markdown( """ Unlike traditional solutions: 1. Complete data privacy through FHE 2. Open-source transparency 3. Cloud-based scanning without local resource consumption 4. Immunity to local malware compromise """ ) gr.Markdown("## 🚀 Go-To-Market Strategy") gr.Markdown("### 🎯 Target Customers") gr.Markdown( """ 1. Enterprise Clients in sensitive industries 2. Cloud Service Providers 3. Cybersecurity Firms 4. Privacy-conscious individuals """ ) gr.Markdown("### 💰 Revenue Model") gr.Markdown( """ 1. Enterprise Licensing 2. Tiered Subscription Plans 3. API Access Fees """ ) gr.Markdown("### 📈 Sales and Marketing Channels") gr.Markdown( """ 1. Direct Enterprise Sales 2. Partnerships with cloud providers and cybersecurity firms 3. Industry events and conferences 4. Content marketing (whitepapers, case studies) """ ) gr.Markdown("## 🏅 Achievements and Roadmap") gr.Markdown("### 🏆 Current Achievements") gr.Markdown( """ - Developed a proof-of-concept multiscanner using the TFHE library - Successfully demonstrated end-to-end process from local encryption to cloud scanning - Implemented a linear classifier for malware detection on encrypted data """ ) gr.Markdown("### 🔮 Future Development") gr.Markdown( """ 1. Develop real-time scanning capabilities 2. Add behavior analysis through encrypted log processing 3. Expand to support a wider range of file types and encryption schemes """ ) demo.launch(share=False)