Upload 11 files

Version_2 - Copy/core/augmentations.py

@@ -0,0 +1,27 @@
+import torch
+from torchvision.transforms import v2
+
+def get_train_transforms(image_size=224):
+    return v2.Compose([
+        v2.RandomResizedCrop(size=(image_size, image_size), antialias=True),
+        v2.RandomHorizontalFlip(p=0.5),
+        v2.RandAugment(num_ops=2, magnitude=9), 
+        v2.ToImage(),
+        v2.ToDtype(torch.float32, scale=True),
+        v2.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
+    ])
+
+def get_eval_transforms(image_size=224):
+    return v2.Compose([
+        v2.Resize(size=(256, 256), antialias=True),
+        v2.CenterCrop(size=(image_size, image_size)),
+        v2.ToImage(),
+        v2.ToDtype(torch.float32, scale=True),
+        v2.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
+    ])
+
+def get_batch_synthetics(num_classes):
+    return v2.RandomChoice([
+        v2.CutMix(num_classes=num_classes, alpha=1.0),
+        v2.MixUp(num_classes=num_classes, alpha=0.8)
+    ])

Version_2 - Copy/core/data_loader.py

@@ -0,0 +1,59 @@
+import torch
+import pandas as pd
+from torch.utils.data import DataLoader, Dataset
+from PIL import Image
+from core.augmentations import get_train_transforms, get_eval_transforms
+
+class PhishingImageDataset(Dataset):
+    def __init__(self, dataframe, transform=None, image_col='image_path', label_col='label'):
+        self.dataframe = dataframe
+        self.transform = transform
+        self.image_col = image_col
+        self.label_col = label_col
+
+    def __len__(self):
+        return len(self.dataframe)
+
+    def __getitem__(self, idx):
+        row = self.dataframe.iloc[idx]
+        img_path = row[self.image_col]
+        label = row[self.label_col]
+        
+        try:
+            image = Image.open(img_path).convert("RGB")
+        except Exception as e:
+            # Fallback to a blank image if file is missing/corrupted
+            print(f"Error loading {img_path}: {e}")
+            image = Image.new('RGB', (224, 224), (0, 0, 0))
+            
+        if self.transform:
+            image = self.transform(image)
+            
+        return image, label
+
+def prepare_dataloaders(legit_csv_path, phishing_csv_path, batch_size=32, image_column_name='image_path'):
+    print(f"Sampling 5,000 rows from {legit_csv_path} and {phishing_csv_path}...")
+    
+    df_legit = pd.read_csv(legit_csv_path).sample(n=5000, random_state=42)
+    df_legit['label'] = 0  # 0: Legit
+    
+    df_phish = pd.read_csv(phishing_csv_path).sample(n=5000, random_state=42)
+    df_phish['label'] = 1  # 1: Phishing
+    
+    df_all = pd.concat([df_legit, df_phish], ignore_index=True)
+    df_all = df_all.sample(frac=1, random_state=42).reset_index(drop=True)
+    
+    # 80/10/10 Split -> 8000 Train, 1000 Val, 1000 Test
+    train_df = df_all.iloc[:8000]
+    val_df = df_all.iloc[8000:9000]
+    test_df = df_all.iloc[9000:]
+    
+    train_ds = PhishingImageDataset(train_df, transform=get_train_transforms(), image_col=image_column_name)
+    val_ds = PhishingImageDataset(val_df, transform=get_eval_transforms(), image_col=image_column_name)
+    test_ds = PhishingImageDataset(test_df, transform=get_eval_transforms(), image_col=image_column_name)
+    
+    train_loader = DataLoader(train_ds, batch_size=batch_size, shuffle=True, num_workers=4)
+    val_loader = DataLoader(val_ds, batch_size=batch_size, shuffle=False, num_workers=4)
+    test_loader = DataLoader(test_ds, batch_size=batch_size, shuffle=False, num_workers=4)
+    
+    return train_loader, val_loader, test_loader, 2

Version_2 - Copy/core/sam.py

@@ -0,0 +1,43 @@
+import torch
+
+class SAM(torch.optim.Optimizer):
+    def __init__(self, params, base_optimizer, rho=0.05, adaptive=False, **kwargs):
+        assert rho >= 0.0, f"Invalid rho, should be non-negative: {rho}"
+        defaults = dict(rho=rho, adaptive=adaptive, **kwargs)
+        super(SAM, self).__init__(params, defaults)
+        self.base_optimizer = base_optimizer(self.param_groups, **kwargs)
+        self.param_groups = self.base_optimizer.param_groups
+        self.defaults.update(self.base_optimizer.defaults)
+
+    @torch.no_grad()
+    def first_step(self, zero_grad=False):
+        grad_norm = self._grad_norm()
+        for group in self.param_groups:
+            scale = group["rho"] / (grad_norm + 1e-12)
+            for p in group["params"]:
+                if p.grad is None: continue
+                self.state[p]["old_p"] = p.data.clone()
+                e_w = (torch.pow(p, 2) if group["adaptive"] else 1.0) * p.grad * scale.to(p)
+                p.add_(e_w)  
+        if zero_grad: self.zero_grad()
+
+    @torch.no_grad()
+    def second_step(self, zero_grad=False):
+        for group in self.param_groups:
+            for p in group["params"]:
+                if p.grad is None: continue
+                p.data = self.state[p]["old_p"]  
+        self.base_optimizer.step()  
+        if zero_grad: self.zero_grad()
+
+    def _grad_norm(self):
+        shared_device = self.param_groups[0]["params"][0].device
+        norm = torch.norm(
+            torch.stack([
+                ((torch.abs(p) if group["adaptive"] else 1.0) * p.grad).norm(p=2).to(shared_device)
+                for group in self.param_groups for p in group["params"]
+                if p.grad is not None
+            ]),
+            p=2
+        )
+        return norm

Version_2 - Copy/inference.py

@@ -0,0 +1,57 @@
+import torch
+import timm
+import ttach as tta
+from PIL import Image
+import os
+
+from core.augmentations import get_eval_transforms
+
+class ProductionAnalyzer:
+    def __init__(self, model_path, num_classes=2):
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self.transform = get_eval_transforms()
+        self.class_names = {0: "Legit", 1: "Phishing"}
+        
+        print("Loading Production Model...")
+        base_model = timm.create_model('convnext_tiny', pretrained=False, num_classes=num_classes)
+        
+        # Load weights and strip the 'module.' prefix caused by AveragedModel saving
+        state_dict = torch.load(model_path, map_location=self.device, weights_only=True)
+        clean_state_dict = {k.replace('module.', ''): v for k, v in state_dict.items()}
+        
+        base_model.load_state_dict(clean_state_dict)
+        base_model.to(self.device)
+        base_model.eval()
+        
+        # Wrap in TTA (Test Time Augmentation)
+        self.model = tta.ClassificationTTAWrapper(
+            base_model, 
+            tta.aliases.five_crop_transform(224, 224)
+        )
+
+    def analyze_user_input(self, image_path):
+        if not os.path.exists(image_path):
+            return "Error: Image file not found."
+            
+        image = Image.open(image_path).convert("RGB")
+        input_tensor = self.transform(image).unsqueeze(0).to(self.device)
+        
+        with torch.no_grad():
+            logits = self.model(input_tensor)
+            probabilities = torch.softmax(logits, dim=1)
+            confidence, predicted_class = torch.max(probabilities, dim=1)
+            
+        class_id = predicted_class.item()
+        return {
+            "prediction": self.class_names[class_id],
+            "class_id": class_id,
+            "confidence_score": f"{confidence.item() * 100:.2f}%"
+        }
+
+if __name__ == "__main__":
+    # Ensure production_convnext_ema.pth exists in the directory before running
+    analyzer = ProductionAnalyzer(model_path="production_convnext_ema.pth")
+    
+    # Example usage:
+    # result = analyzer.analyze_user_input("path_to_screenshot_to_test.jpg")
+    # print(result)

Version_2 - Copy/main.py

@@ -0,0 +1,175 @@
+import os
+import sys
+import time
+import asyncio
+import torch
+import torch.nn.functional as F
+import timm
+from torchvision import transforms
+from PIL import Image
+from playwright.async_api import async_playwright
+
+class ProductionAnalyzer:
+    def __init__(self, model_path="models/production_resnet_ema.pth", model_name="resnet18", num_classes=2):
+        """
+        Initializes the ResNet18 production model on available hardware.
+        """
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        print(f"🖥️ Initializing backend hardware: {self.device}")
+        
+        # 1. Initialize the ResNet18 architecture base
+        print(f"🏗️ Building network architecture: {model_name}...")
+        self.model = timm.create_model(model_name, pretrained=False, num_classes=num_classes)
+        
+        # 2. Safety check for the weights file
+        if not os.path.exists(model_path):
+            raise FileNotFoundError(f"❌ Could not find weight file at: {model_path}\n"
+                                    f"Please ensure it is placed inside the 'models' folder.")
+            
+        print(f"📥 Loading ResNet18 weights from {model_path}...")
+        state_dict = torch.load(model_path, map_location=self.device, weights_only=True)
+        
+        # 3. 🛠️ FIXED: Strip wrapper prefixes and safely drop training metadata keys
+        clean_state_dict = {}
+        for key, value in state_dict.items():
+            if key == "n_averaged":
+                continue # Skip the training counter metadata so PyTorch doesn't throw an error
+                
+            clean_key = key.replace('module.', '')
+            clean_state_dict[clean_key] = value
+            
+        self.model.load_state_dict(clean_state_dict)
+        
+        # 4. Lock the model for evaluation mode
+        self.model.eval()
+        self.model.to(self.device)
+        print("✅ ResNet18 Engine successfully loaded and locked for inference.")
+        
+        # 5. Define standard normalizations expected by ResNet18
+        self.transform = transforms.Compose([
+            transforms.Resize((224, 224)),
+            transforms.ToTensor(),
+            transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
+        ])
+        
+        # Alphabetical class mapping (0: Legit, 1: Phishing)
+        self.class_names = ["Legit", "Phishing"]
+
+    def analyze_image(self, image_path):
+        """
+        Feeds the captured screenshot into the ResNet18 neural network.
+        """
+        try:
+            image = Image.open(image_path).convert('RGB')
+            input_tensor = self.transform(image).unsqueeze(0).to(self.device)
+            
+            with torch.no_grad():
+                logits = self.model(input_tensor)
+                probabilities = F.softmax(logits[0], dim=0)
+                confidence, predicted_idx = torch.max(probabilities, 0)
+                
+            return {
+                "prediction": self.class_names[predicted_idx.item()],
+                "confidence": f"{confidence.item() * 100:.2f}%"
+            }
+        except Exception as e:
+            return {"error": f"Model inference failed: {str(e)}"}
+
+
+async def capture_screenshot(url, output_path="temp_inference.png"):
+    """
+    Launches a headless browser to safely capture a screenshot of the live URL.
+    """
+    if not url.startswith('http://') and not url.startswith('https://'):
+        url = 'https://' + url
+        
+    print(f"🌐 Navigating to: {url} ...")
+    
+    async with async_playwright() as p:
+        browser = await p.chromium.launch(headless=True)
+        context = await browser.new_context(
+            viewport={'width': 1280, 'height': 720},
+            ignore_https_errors=True, # Bypasses broken SSL certifications on malicious sites
+            accept_downloads=False
+        )
+        
+        page = await context.new_page()
+        try:
+            # 12-second timeout to handle slow/malicious servers
+            await page.goto(url, timeout=12000, wait_until='domcontentloaded')
+            await asyncio.sleep(1) # Brief pause to let visual components load completely
+            await page.screenshot(path=output_path)
+            return output_path
+        except Exception as e:
+            print(f"❌ Failed to reach or capture the website: {e}")
+            return None
+        finally:
+            await page.close()
+            await browser.close()
+
+
+async def main():
+    print("=" * 45)
+    print("🛡️ CHIMERA 2.0 LIVE URL DETECTOR ENGINE")
+    print("=" * 45)
+    
+    try:
+        analyzer = ProductionAnalyzer()
+    except Exception as e:
+        print(e)
+        return
+
+    TEMP_IMG = "temp_inference.png"
+    
+    try:
+        while True:
+            print("\n" + "-" * 45)
+            url_input = input("🔗 Enter URL to inspect (or type 'exit' to quit): ").strip()
+            
+            if url_input.lower() == 'exit':
+                print("Shutting down engine...")
+                break
+            if not url_input:
+                continue
+                
+            start_time = time.time()
+            
+            # Step 1: Capture screenshot via Playwright
+            screenshot_file = await capture_screenshot(url_input, TEMP_IMG)
+            
+            if screenshot_file and os.path.exists(screenshot_file):
+                # Step 2: Pass screenshot into ResNet18
+                print("🔍 Running Deep Learning Visual Inspection...")
+                result = analyzer.analyze_image(screenshot_file)
+                
+                total_time = time.time() - start_time
+                
+                # Step 3: Output results safely
+                print("\n" + "=" * 35)
+                print("📊 LIVE DETECTION REPORT")
+                print("=" * 35)
+                if "error" in result:
+                    print(f"Result: {result['error']}")
+                else:
+                    status_prefix = "🚨 ALERT!!" if result['prediction'] == "Phishing" else "✅ CLEAR:"
+                    print(f"Verdict    : {status_prefix} {result['prediction']}")
+                    print(f"Confidence : {result['confidence']}")
+                    print(f"Total Time : {total_time:.2f} seconds")
+                print("=" * 35)
+                
+                if os.path.exists(TEMP_IMG):
+                    os.remove(TEMP_IMG)
+            else:
+                print("❌ Inspection aborted. Visual fingerprint could not be gathered.")
+                
+    except KeyboardInterrupt:
+        print("\nExiting execution gracefully...")
+    finally:
+        if os.path.exists(TEMP_IMG):
+            os.remove(TEMP_IMG)
+
+if __name__ == "__main__":
+    if sys.platform == 'win32':
+        asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
+        
+    asyncio.run(main())

Version_2 - Copy/models/production_resnet_ema.pth

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:75e79d34179ef20f3fd3914e18c6867f4dfd356f8edd2b483eb5ba74779c7796
+size 44794007