predict_url.py

"""
URLAZ Phishing Predictor CLI (P0 IDN Bias Fix Engine Enabled)
============================================================
Run interactive URL classification using:
- Tier 1: High-Confidence Allowlist Pre-Filter (Clean Official Domains)
- Tier 2: 38 Structural Feature LightGBM Model (P0 IDN Bias Fix Engine + UTS #39 Confusable Skeleton)
- Operational Threshold t = 0.95 (Exact FPR < 0.09%, Production Precision 52.46%)
"""

import os
import sys
import json
import joblib
import numpy as np

# OPSEC Clean Dynamic Relative Paths
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
BASE_DIR = os.path.dirname(SCRIPT_DIR)
DATA_DIR = os.path.join(BASE_DIR, "data")
sys.path.append(os.path.join(BASE_DIR, "scripts"))

from train_production_model_v6 import (
    extract_features_v5,
    check_allowlist_bypass,
    extract_domain_from_url,
    get_registered_domain,
    detect_homoglyph_typosquatting,
    decode_punycode_and_to_skeleton,
    FEATURE_NAMES_V5,
    FULL_ALLOWLIST_DOMAINS,
    AZ_BRAND_PATTERNS
)

DIST_MODEL_PATH = os.path.join(BASE_DIR, "data_dist", "urlaz_phishing_detector.joblib")
DATA_MODEL_PATH = os.path.join(DATA_DIR, "az_phish_model_v6.joblib")

MODEL_PATH = DIST_MODEL_PATH if os.path.exists(DIST_MODEL_PATH) else DATA_MODEL_PATH

if not os.path.exists(MODEL_PATH):
    raise FileNotFoundError(f"Model weights file not found at {MODEL_PATH}")

clf = joblib.load(MODEL_PATH)
OPERATIONAL_THRESHOLD = 0.95

def analyze_url(url: str) -> dict:
    url_clean = str(url).strip()
    domain = extract_domain_from_url(url_clean)
    registered_domain = get_registered_domain(domain)
    
    # Step 1: Check Allowlist Bypass
    is_allowlisted = check_allowlist_bypass(url_clean)
    
    if is_allowlisted:
        risk_prob = 0.0
        verdict = "BENIGN / SAFE 🟢"
        decision_path = "Tier 1 Allowlist Pre-Filter (0ms Latency Bypass)"
        key_signals = [f"Registered Domain '{registered_domain}' is in High-Confidence Allowlist"]
    else:
        # Step 2: Tier 2 ML Classifier (38 Features)
        features = extract_features_v5(url_clean)
        risk_prob = float(clf.predict_proba([features])[0][1])
        
        if risk_prob >= OPERATIONAL_THRESHOLD:
            verdict = "PHISHING / HIGH RISK 🔴"
        elif risk_prob >= 0.50:
            verdict = "SUSPICIOUS / MEDIUM RISK 🟡"
        else:
            verdict = "BENIGN / LOW RISK 🟢"
            
        decision_path = f"Tier 2 LightGBM Classifier (Operational Threshold = {OPERATIONAL_THRESHOLD})"
        
        # Identify key signals cleanly
        key_signals = []
        skeleton_dom = decode_punycode_and_to_skeleton(domain)
        brand_matches = [b for b in AZ_BRAND_PATTERNS if b in skeleton_dom]
        if brand_matches:
            key_signals.append(f"Target Brand Keyword Match: {brand_matches}")
            
        # Check Homoglyph / Typosquatting / Leetspeak Signal
        typo_flag, typo_dist, is_homoglyph = detect_homoglyph_typosquatting(domain)
        if is_homoglyph:
            key_signals.append(f"UTS #39 Homoglyph / IDN Punycode Attack Detected in '{domain}' -> skeleton: '{skeleton_dom}'")
        elif typo_flag > 0 and 1 <= typo_dist <= 2:
            key_signals.append(f"Leetspeak / Typosquatting Impersonation Detected in '{domain}' (Edit Dist = {typo_dist})")
            
        if domain != registered_domain and any(b in skeleton_dom for b in AZ_BRAND_PATTERNS):
            key_signals.append(f"Subdomain Brand Spoofing detected in '{domain}'")
        if any(domain.endswith(t) for t in ['.xyz', '.top', '.site', '.cc', '.fun', '.online', '.cfd', '.vip']):
            key_signals.append(f"High-Risk TLD observed: '{domain.split('.')[-1]}'")
        if len(url_clean) > 75:
            key_signals.append(f"Suspiciously long URL ({len(url_clean)} chars)")
            
    risk_score = int(round(risk_prob * 100))
    
    return {
        "url": url_clean,
        "domain": domain,
        "registered_domain": registered_domain,
        "verdict": verdict,
        "risk_score": risk_score,
        "risk_probability": round(risk_prob, 4),
        "decision_path": decision_path,
        "key_signals": key_signals if key_signals else ["Normal structural patterns observed"]
    }

def print_result(res: dict):
    print("\n" + "═" * 70)
    print(f" 🔍 URL ANALYSIS REPORT: {res['url']}")
    print("═" * 70)
    print(f"  VERDICT             : {res['verdict']}")
    print(f"  RISK SCORE (0-100)  : {res['risk_score']} / 100  (Probability: {res['risk_probability']*100:.2f}%)")
    print(f"  DECISION PATH       : {res['decision_path']}")
    print(f"  DOMAIN INFO         : {res['domain']} (Registered: {res['registered_domain']})")
    print("  KEY DETECTED SIGNALS:")
    for sig in res['key_signals']:
        print(f"    • {sig}")
    print("═" * 70 + "\n")

if __name__ == "__main__":
    if len(sys.argv) > 1:
        test_url = sys.argv[1]
        res = analyze_url(test_url)
        print_result(res)
    else:
        print("URLAZ Phishing Predictor CLI Ready. Enter a URL to analyze (or 'q' to quit):")
        while True:
            try:
                inp = input("\nURL > ").strip()
                if not inp or inp.lower() in ('q', 'exit', 'quit'):
                    break
                res = analyze_url(inp)
                print_result(res)
            except (KeyboardInterrupt, EOFError):
                break