Delete Version_3

Version_3/app.py

@@ -1,119 +0,0 @@
-import os
-import sys
-import time
-from fastapi import FastAPI, HTTPException
-from pydantic import BaseModel
-
-# =====================================================================
-# SAFE PIPELINE IMPORT MATRIX
-# =====================================================================
-try:
-    from pipeline.tier1_fast_filter import check_fast_filter
-    from pipeline.tier2_extraction import sanitize_text, extract_metadata_flags
-    from pipeline.tier3_inference import neural_engine
-    from pipeline.tier4_scoring import calculate_final_verdict
-except ModuleNotFoundError:
-    from Version_3.pipeline.tier1_fast_filter import check_fast_filter
-    from Version_3.pipeline.tier2_extraction import sanitize_text, extract_metadata_flags
-    from Version_3.pipeline.tier3_inference import neural_engine
-    from Version_3.pipeline.tier4_scoring import calculate_final_verdict
-
-# =====================================================================
-# INITIALIZE APP INSTANCE
-# =====================================================================
-app = FastAPI(title="Real-Time NLP Phishing Defender")
-
-class MessagePayload(BaseModel):
-    message_id: str
-    sender_domain: str
-    display_name: str
-    raw_text: str
-
-# =====================================================================
-# UNIFIED ROUTE ENDPOINT
-# =====================================================================
-@app.post("/analyze")
-async def analyze_message(payload: MessagePayload):
-    start_time = time.perf_counter()
-    
-    # --- TIER 1: Fast Filter ---
-    t1_result = check_fast_filter(payload.sender_domain, payload.raw_text)
-    if t1_result["bypassed"]:
-        latency_ms = (time.perf_counter() - start_time) * 1000
-        return {
-            "message_id": payload.message_id,
-            "verdict": t1_result["verdict"],
-            "latency_ms": round(latency_ms, 2),
-            "bypassed_at_tier": 1
-        }
-        
-    # --- TIER 2: Extraction & Sanitization ---
-    clean_text = sanitize_text(payload.raw_text)
-    flags = extract_metadata_flags(payload.sender_domain, payload.display_name, clean_text)
-    
-    # --- TIER 3: Neural Inference ---
-    nlp_prob = neural_engine.predict_phishing_prob(clean_text)
-    
-    # --- TIER 4: Scoring Matrix ---
-    final_result = calculate_final_verdict(nlp_prob, flags)
-    
-    latency_ms = (time.perf_counter() - start_time) * 1000
-    
-    return {
-        "message_id": payload.message_id,
-        "verdict": final_result["verdict"],
-        "risk_score": final_result["risk_score"],
-        "details": final_result["breakdown"],
-        "latency_ms": round(latency_ms, 2)
-    }
-
-# =====================================================================
-# STANDALONE LOCAL EXECUTION GATEWAY
-# =====================================================================
-if __name__ == "__main__":
-    # Check if we explicitly want to run in local interactive mode
-    if len(sys.argv) > 1 and sys.argv[1] == "--cli":
-        print("\n" + "="*50)
-        print("  CHIMERA 2.0 LOCAL TERMINAL ENGINE LOADED  ")
-        print("  (Press Ctrl+C at any time to exit)        ")
-        print("="*50 + "\n")
-        
-        while True:
-            try:
-                domain = input("1. Enter Sender Domain: ").strip()
-                name = input("2. Enter Display Name: ").strip()
-                text = input("3. Enter Email Content Text: ").strip()
-                
-                if not text:
-                    print("\nError: Text cannot be empty.\n")
-                    continue
-                    
-                start = time.perf_counter()
-                
-                # Run the exact pipeline logic defined in your endpoint
-                t1 = check_fast_filter(domain, text)
-                if t1["bypassed"]:
-                    latency = (time.perf_counter() - start) * 1000
-                    print(f"\n[🛑 TIER 1 HIT] Verdict: {t1['verdict']} | Latency: {round(latency, 2)} ms\n" + "-"*50)
-                    continue
-                    
-                clean = sanitize_text(text)
-                flags = extract_metadata_flags(domain, name, clean)
-                nlp_prob = neural_engine.predict_phishing_prob(clean)
-                final = calculate_final_verdict(nlp_prob, flags)
-                
-                latency = (time.perf_counter() - start) * 1000
-                print(f"\n" + "═"*15 + " ANALYSIS RESULT " + "═"*15)
-                print(f"VERDICT:     {final['verdict']}")
-                print(f"RISK SCORE:  {final['risk_score']}")
-                print(f"LATENCY:     {round(latency, 2)} ms")
-                print(f"DETAILS:     {final['breakdown']}")
-                print("═"*47 + "\n")
-                
-            except KeyboardInterrupt:
-                print("\nExiting interactive mode.")
-                break
-    else:
-        # Fallback default behavior to run the web server if no flag is provided
-        import uvicorn
-        uvicorn.run(app, host="0.0.0.0", port=8000)

Version_3/config.py

@@ -1,22 +0,0 @@
-import os
-
-class Config:
-    # --- Dynamic Path Resolution for Hugging Face ---
-    # This automatically finds the absolute path to the Version_4 folder
-    BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-    
-    # Model Paths
-    MODEL_PATH = os.getenv(
-        "ONNX_MODEL_PATH", 
-        os.path.join(BASE_DIR, "models", "model_quantized.onnx")
-    )
-    TOKENIZER_PATH = os.getenv("TOKENIZER_PATH", "microsoft/deberta-v3-base")
-    
-    # Tier 4 Decision Matrix Weights
-    W_NLP = 0.70
-    W_META = 0.30
-    THRESHOLD = 0.40
-    
-    # Tier 1 Cache (Simulated with local sets for this pipeline)
-    KNOWN_SAFE_DOMAINS = {"google.com", "microsoft.com", "apple.com", "internal-corp.com"}
-    KNOWN_MALICIOUS_HASHES = {"e99a18c428cb38d5f260853678922e03"} # Example MD5s

Version_3/models/model_quantized.onnx

@@ -1,3 +0,0 @@
-version https://git-lfs.github.com/spec/v1
-oid sha256:07eb49da681035631002cf959d0fb34e9c9efd5016d884ff1170228d841d47fb
-size 244570129

Version_3/pipeline/tier1_fast_filter.py

@@ -1,18 +0,0 @@
-import hashlib
-from config import Config
-
-def compute_hash(text: str) -> str:
-    return hashlib.md5(text.encode('utf-8')).hexdigest()
-
-def check_fast_filter(sender_domain: str, raw_text: str):
-    """
-    Tier 1: Deterministic $O(1)$ lookup. Returns a definitive verdict instantly if matched.
-    """
-    if sender_domain.lower() in Config.KNOWN_SAFE_DOMAINS:
-        return {"bypassed": True, "verdict": "SAFE", "reason": "Trusted Domain"}
-    
-    msg_hash = compute_hash(raw_text)
-    if msg_hash in Config.KNOWN_MALICIOUS_HASHES:
-        return {"bypassed": True, "verdict": "PHISHING", "reason": "Known Threat Signature"}
-        
-    return {"bypassed": False}

Version_3/pipeline/tier2_extraction.py

@@ -1,36 +0,0 @@
-import re
-import unicodedata
-
-# Common obfuscation characters (e.g., zero-width spaces)
-OBFUSCATION_CHARS = r'[\u200B\u200C\u200D\uFEFF]'
-URL_SHORTENERS = r'(bit\.ly|tinyurl\.com|t\.co|ow\.ly|is\.gd)'
-URGENCY_KEYWORDS = r'(immediate action|account suspended|verify now|unauthorized login|final warning)'
-
-def sanitize_text(text: str) -> str:
-    """Strips zero-width characters and normalizes lookalike Unicode."""
-    clean_text = re.sub(OBFUSCATION_CHARS, '', text)
-    clean_text = unicodedata.normalize('NFKC', clean_text)
-    return clean_text
-
-def extract_metadata_flags(sender_domain: str, display_name: str, clean_text: str) -> int:
-    """
-    Tier 2: Extracts behavioral flags. Each triggered rule adds to the flag count.
-    """
-    flags = 0
-    text_lower = clean_text.lower()
-    
-    # Flag 1: Domain / Display Name mismatch (e.g., "PayPal" sending from "support@cheap-shoes.xyz")
-    if display_name and display_name.lower() != sender_domain.lower():
-        # A naive check; in production, use Levenshtein distance against known brands
-        if "paypal" in display_name.lower() and "paypal.com" not in sender_domain.lower():
-            flags += 1
-
-    # Flag 2: Presence of URL shorteners
-    if re.search(URL_SHORTENERS, text_lower):
-        flags += 1
-        
-    # Flag 3: Urgency / Threat-to-Action density
-    if re.search(URGENCY_KEYWORDS, text_lower):
-        flags += 1
-        
-    return flags

Version_3/pipeline/tier3_inference.py

@@ -1,59 +0,0 @@
-import onnxruntime as ort
-import numpy as np
-from transformers import AutoTokenizer
-from config import Config
-
-class NeuralEngine:
-    def __init__(self):
-        # Load HuggingFace tokenizer (Fast tokenizer written in Rust for low latency)
-        self.tokenizer = AutoTokenizer.from_pretrained(Config.TOKENIZER_PATH)
-        
-        # Initialize ONNX CPU runtime for INT8 quantized model
-        # intra_op_num_threads limits thread spinning to hit the 15ms target consistently
-        sess_options = ort.SessionOptions()
-        sess_options.intra_op_num_threads = 2 
-        sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
-        
-        try:
-            self.session = ort.InferenceSession(Config.MODEL_PATH, sess_options, providers=['CPUExecutionProvider'])
-            self.model_loaded = True
-        except Exception as e:
-            print(f"Warning: ONNX model not found at {Config.MODEL_PATH}. Running in mock mode.")
-            self.model_loaded = False
-
-    def predict_phishing_prob(self, text: str) -> float:
-        """
-        Tier 3: Transformer execution. Uses dynamic padding to minimize matrix sizes.
-        """
-        # Dynamic Padding: only pad up to the longest sequence in this specific batch (1 in this case)
-        inputs = self.tokenizer(
-            text, 
-            return_tensors="np", 
-            truncation=True, 
-            max_length=512, 
-            padding=False 
-        )
-        
-        if not self.model_loaded:
-            # Mock output for testing without the actual heavy model file
-            return 0.85 if "password" in text.lower() else 0.05
-            
-        ort_inputs = {
-            "input_ids": inputs["input_ids"].astype(np.int64),
-            "attention_mask": inputs["attention_mask"].astype(np.int64)
-        }
-        
-        # Run forward pass
-        ort_outs = self.session.run(None, ort_inputs)
-        
-        # Assuming binary classification where index 1 is 'Phishing'
-        logits = ort_outs[0][0]
-        
-        # Softmax to get probability
-        exp_logits = np.exp(logits - np.max(logits))
-        probabilities = exp_logits / exp_logits.sum()
-        
-        return float(probabilities[1])
-        
-# Singleton instantiation to keep model loaded in memory
-neural_engine = NeuralEngine()

Version_3/pipeline/tier4_scoring.py

@@ -1,30 +0,0 @@
-from config import Config
-
-def calculate_final_verdict(nlp_prob, flags):
-    # 'flags' is an integer representing the total count of suspicious flags
-    base_score = nlp_prob * 100
-    
-    # Add a risk penalty for every flag found (e.g., 10 points per flag)
-    penalty = flags * 10 
-    
-    actual_calculated_score = min(100, base_score + penalty)
-    
-    if actual_calculated_score >= 40:
-        actual_verdict_string = "Phishing"
-    else:
-        actual_verdict_string = "Safe"
-
-    # THE NEW BREAKDOWN DICTIONARY
-    detailed_breakdown = {
-        "language_analysis": f"NLP matched phishing patterns with {round(nlp_prob * 100, 2)}% confidence." if nlp_prob >= 0.4 else "Text appears conversational and safe.",
-        
-        "link_analysis": f"Detected {flags} suspicious metadata/link flags." if flags > 0 else "Metadata and links appear clean.",
-        
-        "sender_verification": "Suspicious metadata detected." if flags > 0 else "Sender identity verified and clean."
-    }
-    
-    return {
-        "verdict": actual_verdict_string, 
-        "risk_score": actual_calculated_score, 
-        "breakdown": detailed_breakdown 
-    }