Upload 10 files

Version_5/app.py

@@ -0,0 +1,90 @@
+import time
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel, Field
+
+# Absolute path imports mapped to your Mega-Server setup
+from Version_5.src.dom_scraper import extract_dom_features
+from Version_5.src.sub_agents import (
+    agent_url_analyst, 
+    agent_html_structure, 
+    agent_content_semantics, 
+    agent_brand_impersonation
+)
+from Version_5.src.orchestrator import evaluate_consensus, run_judge
+
+router = APIRouter()
+
+# Define API request payload structure
+class ThreatAnalysisRequest(BaseModel):
+    url: str = Field(..., description="Target landing page URL to analyze", example="http://example.com")
+    sender: str = Field(..., description="Alleged sender address header", example="security@paypal.com")
+    email_body: str = Field(..., description="Full text/body payload of the incoming message")
+
+@router.post("/predict")
+async def analyze_payload_endpoint(payload: ThreatAnalysisRequest):
+    url = payload.url.strip()
+    sender = payload.sender.strip()
+    email_body = payload.email_body.strip()
+    
+    if not url and not email_body:
+        raise HTTPException(
+            status_code=400, 
+            detail="Structural requirement breach: Provide at least a validation URL or a message body."
+        )
+        
+    try:
+        start_time = time.perf_counter()
+        
+        # Tier 1: Headless Chromium Capture
+        dom_data = extract_dom_features(url)
+        
+        # Tier 2: Multi-Specialist Forensic Panel
+        reports = {
+            "URL_Agent": agent_url_analyst(url),
+            "HTML_Agent": agent_html_structure(dom_data),
+            "Content_Agent": agent_content_semantics(email_body),
+            "Brand_Agent": agent_brand_impersonation(email_body, sender)
+        }
+        
+        # Tier 3: Core Consensus Evaluator
+        consensus_victory = evaluate_consensus(reports)
+        
+        if consensus_victory:
+            # Handle standard serialization if agents return Pydantic objects or plain strings
+            final_verdict = {
+                "verdict": reports["URL_Agent"].claim if hasattr(reports["URL_Agent"], "claim") else str(reports["URL_Agent"]),
+                "confidence_score": reports["URL_Agent"].confidence if hasattr(reports["URL_Agent"], "confidence") else 1.0,
+                "justification": "Bypassed judicial review due to absolute sub-agent unanimity across forensics."
+            }
+        else:
+            # Tier 4: Judicial Override via Groq Cloud API
+            reports_str = "\n".join([
+                f"[{name}]\n{r.model_dump_json(indent=2) if hasattr(r, 'model_dump_json') else str(r)}" 
+                for name, r in reports.items()
+            ])
+            raw_data = f"Target URL: {url}\nTarget Sender: {sender}\nBody: {email_body}"
+            
+            judge_verdict = run_judge(reports_str, raw_data)
+            final_verdict = judge_verdict.model_dump() if hasattr(judge_verdict, "model_dump") else judge_verdict
+            
+        latency_ms = (time.perf_counter() - start_time) * 1000
+        
+        # Safe response serialization
+        serializable_reports = {}
+        for name, report in reports.items():
+            serializable_reports[name] = report.model_dump() if hasattr(report, "model_dump") else str(report)
+
+        return {
+            "target_url": url,
+            "target_sender": sender,
+            "consensus_reached": consensus_victory,
+            "latency_ms": round(latency_ms, 2),
+            "sub_agent_claims": serializable_reports,
+            "final_evaluation": final_verdict
+        }
+
+    except Exception as e:
+        raise HTTPException(
+            status_code=500, 
+            detail=f"Internal agent execution lifecycle crash: {str(e)}"
+        )

Version_5/src/dom_scraper.py

@@ -0,0 +1,28 @@
+from playwright.sync_api import sync_playwright
+import json
+
+def extract_dom_features(url: str) -> str:
+    if not url.startswith("http"):
+        return json.dumps({"error": "No valid URL provided."})
+        
+    try:
+        with sync_playwright() as p:
+            browser = p.chromium.launch(headless=True)
+            page = browser.new_page()
+            # 10-second timeout to prevent stalling on dead phishing links
+            page.goto(url, timeout=10000) 
+            
+            # Extract where login forms actually send data
+            forms = page.locator("form").evaluate_all(
+                "elements => elements.map(e => ({ action: e.action, method: e.method }))"
+            )
+            
+            # Extract invisible tracking/token fields
+            hidden_inputs = page.locator("input[type='hidden']").evaluate_all(
+                "elements => elements.map(e => ({ name: e.name, value: e.value }))"
+            )
+            
+            browser.close()
+            return json.dumps({"forms": forms, "hidden_inputs": hidden_inputs})
+    except Exception as e:
+        return json.dumps({"scraper_error": str(e)})

Version_5/src/orchestrator.py

@@ -0,0 +1,38 @@
+import os
+from groq import Groq
+from typing import Dict
+from src.schemas import AgentReport, JudgeVerdict
+
+client = Groq(api_key=os.environ.get("GROQ_API_KEY"))
+
+def evaluate_consensus(reports: Dict[str, AgentReport]) -> bool:
+    claims = [r.claim for r in reports.values()]
+    confidences = [r.confidence for r in reports.values()]
+    
+    all_same = len(set(claims)) == 1
+    high_conf = all(c >= 0.85 for c in confidences)
+    
+    return all_same and high_conf
+
+def run_judge(reports_summary: str, raw_data: str) -> JudgeVerdict:
+    system_prompt = (
+        "You are the presiding Judge Magistrate in an Enterprise Cybersecurity Matrix.\n"
+        "Your subordinate agents have provided split panel reports.\n"
+        "Respond strictly in valid JSON matching this schema:\n"
+        "{\"final_verdict\": \"CRITICAL_THREAT\"|\"SAFE\"|\"MANUAL_REVIEW_REQUIRED\", \"risk_score\": float, \"verdict_justification\": string}\n\n"
+        "DECISION ARCHITECTURE RULES:\n"
+        "1. If ANY analyst engine raises a 'PHISHING' claim with confidence >= 0.85, force 'CRITICAL_THREAT'.\n"
+        "2. Avoid label contradictions: Never set final_verdict to SAFE if your textual argument implies a threat pattern."
+    )
+    
+    # We leverage the powerful 70B cloud model for premium reasoning accuracy
+    response = client.chat.completions.create(
+        model="llama-3.3-70b-versatile",
+        messages=[
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": f"RAW INPUTS:\n{raw_data}\n\nREPORTS:\n{reports_summary}"}
+        ],
+        response_format={"type": "json_object"},
+        temperature=0.0
+    )
+    return JudgeVerdict.model_validate_json(response.choices[0].message.content)

Version_5/src/schemas.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel, Field
+from typing import List, Literal
+
+class AgentReport(BaseModel):
+    claim: Literal["PHISHING", "LEGITIMATE", "AMBIGUOUS"]
+    confidence: float = Field(ge=0.0, le=1.0, description="Confidence score from 0.0 to 1.0")
+    evidence: List[str] = Field(description="Bullet points of forensic evidence found")
+
+class JudgeVerdict(BaseModel):
+    final_verdict: Literal["CRITICAL_THREAT", "SAFE", "MANUAL_REVIEW_REQUIRED"]
+    risk_score: float = Field(ge=0.0, le=1.0)
+    verdict_justification: str = Field(description="Detailed explanation of the final decision")

Version_5/src/sub_agents.py

@@ -0,0 +1,52 @@
+import os
+from groq import Groq
+from src.schemas import AgentReport
+
+# Initialize the Groq Cloud Client
+client = Groq(api_key=os.environ.get("GROQ_API_KEY"))
+
+def run_agent(role_prompt: str, target_data: str) -> AgentReport:
+    # We call Groq's hosted 8B infrastructure instead of local resources
+    response = client.chat.completions.create(
+        model="llama-3.1-8b-instant",
+        messages=[
+            {"role": "system", "content": role_prompt},
+            {"role": "user", "content": target_data}
+        ],
+        # Groq enforces JSON mode via this parameter block
+        response_format={"type": "json_object"},
+        temperature=0.0
+    )
+    return AgentReport.model_validate_json(response.choices[0].message.content)
+
+def agent_url_analyst(url: str) -> AgentReport:
+    prompt = (
+        "You are a URL Shadows Analyst. Respond strictly in valid JSON matching this schema:\n"
+        "{\"claim\": \"PHISHING\"|\"LEGITIMATE\"|\"AMBIGUOUS\", \"confidence\": float, \"evidence\": [string]}\n"
+        "Analyze the structure for structural entropy, unusual domains, and brand keywords."
+    )
+    return run_agent(prompt, f"URL to analyze: {url}")
+
+def agent_html_structure(dom_json: str) -> AgentReport:
+    prompt = (
+        "You are an HTML Code Analyst. Respond strictly in valid JSON matching this schema:\n"
+        "{\"claim\": \"PHISHING\"|\"LEGITIMATE\"|\"AMBIGUOUS\", \"confidence\": float, \"evidence\": [string]}\n"
+        "Analyze these DOM elements. Flag external tracking inputs or forms posting to alternative servers."
+    )
+    return run_agent(prompt, f"DOM Data: {dom_json}")
+
+def agent_content_semantics(email_body: str) -> AgentReport:
+    prompt = (
+        "You are a Phishing Copywriter Critic. Respond strictly in valid JSON matching this schema:\n"
+        "{\"claim\": \"PHISHING\"|\"LEGITIMATE\"|\"AMBIGUOUS\", \"confidence\": float, \"evidence\": [string]}\n"
+        "Extract semantic compliance anomalies, high emotional coercion markers, and urgency loops."
+    )
+    return run_agent(prompt, f"Email Body: {email_body}")
+
+def agent_brand_impersonation(email_body: str, sender: str) -> AgentReport:
+    prompt = (
+        "You are an Identity Protection Agent. Respond strictly in valid JSON matching this schema:\n"
+        "{\"claim\": \"PHISHING\"|\"LEGITIMATE\"|\"AMBIGUOUS\", \"confidence\": float, \"evidence\": [string]}\n"
+        "Flag mismatched target namespaces where sender domains do not match corporate identifiers."
+    )
+    return run_agent(prompt, f"Sender: {sender}\nBody: {email_body}")