streamlit_app.py

"""
Utility functions for Domain Threat Intelligence Dashboard
"""

import streamlit as st
import pandas as pd
import random
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
import hashlib

# Mock data generators for PoC demonstration

def generate_verdict() -> Dict:
    """Generate a random verdict with confidence"""
    verdicts = ["Malicious", "Suspicious", "Benign", "Unknown"]
    weights = [0.15, 0.25, 0.50, 0.10]
    verdict = random.choices(verdicts, weights=weights)[0]
    
    confidence = {
        "Malicious": random.randint(75, 99),
        "Suspicious": random.randint(50, 74),
        "Benign": random.randint(80, 99),
        "Unknown": random.randint(0, 49)
    }[verdict]
    
    reasoning = {
        "Malicious": [
            "Confirmed command-and-control infrastructure for APT28",
            "Associated with multiple malware distribution campaigns",
            "Linked to documented phishing operations",
            "Identified as hosting exploit kits and ransomware payloads"
        ],
        "Suspicious": [
            "Recently registered with privacy-protected registration",
            "Hosting content with ambiguous intent",
            "Traffic patterns inconsistent with normal web activity",
            "Whois data matches known bulletproof hosting patterns"
        ],
        "Benign": [
            "Established business presence with consistent reputation",
            "No adverse findings across major security feeds",
            "Legitimate e-commerce infrastructure",
            "Verified organization with valid certificates"
        ],
        "Unknown": [
            "Insufficient data for comprehensive analysis",
            "Recently registered domain with no historical context",
            "Limited external references available",
            "Analysis ongoing - awaiting additional signals"
        ]
    }
    
    return {
        "label": verdict,
        "confidence": confidence,
        "reasoning": random.choice(reasoning[verdict]),
        "last_updated": datetime.now() - timedelta(hours=random.randint(0, 48))
    }

def generate_intelligence_tags(verdict: str) -> List[Dict]:
    """Generate intelligence tags based on verdict"""
    tag_pool = {
        "Malicious": [
            {"name": "C2", "category": "Threat Type", "severity": "critical"},
            {"name": "APT-linked", "category": "Attribution", "severity": "critical"},
            {"name": "Malware Infrastructure", "category": "Category", "severity": "high"},
            {"name": "Phishing Kit", "category": "Threat Type", "severity": "high"},
            {"name": "Exploit Kit", "category": "Threat Type", "severity": "high"}
        ],
        "Suspicious": [
            {"name": "Recently Registered", "category": "Registration", "severity": "medium"},
            {"name": "Privacy Protected", "category": "Registration", "severity": "medium"},
            {"name": "Bulletproof Hosting", "category": "Infrastructure", "severity": "medium"},
            {"name": "Fast Flux", "category": "Infrastructure", "severity": "medium"}
        ],
        "Benign": [
            {"name": "Legitimate Business", "category": "Classification", "severity": "low"},
            {"name": "E-commerce", "category": "Category", "severity": "low"},
            {"name": "Verified Publisher", "category": "Trust", "severity": "low"}
        ],
        "Unknown": [
            {"name": "Limited Data", "category": "Context", "severity": "low"},
            {"name": "New Domain", "category": "Registration", "severity": "low"}
        ]
    }
    
    available_tags = tag_pool.get(verdict, tag_pool["Unknown"])
    num_tags = random.randint(2, min(4, len(available_tags)))
    selected = random.sample(available_tags, num_tags)
    
    # Add some common tags
    first_seen = datetime.now() - timedelta(days=random.randint(1, 365))
    last_seen = datetime.now() - timedelta(days=random.randint(0, 30))
    
    for tag in selected:
        tag["first_seen"] = first_seen.strftime("%Y-%m-%d")
        tag["last_seen"] = last_seen.strftime("%Y-%m-%d")
    
    return selected

def generate_source_verdicts() -> List[Dict]:
    """Generate verdicts from different intelligence sources"""
    sources = [
        {"name": "Internal Dataset", "type": "Reputation", "weight": 0.9},
        {"name": "URL Scanner", "type": "Dynamic Analysis", "weight": 0.8},
        {"name": "Reputation Feed", "type": "Threat Intel", "weight": 0.7},
        {"name": "Passive DNS", "type": "Infrastructure", "weight": 0.6},
        {"name": "Web Search", "type": "OSINT", "weight": 0.5},
        {"name": "WHOIS Database", "type": "Registration", "weight": 0.4},
        {"name": "Certificate Transparency", "type": "Infrastructure", "weight": 0.5},
        {"name": "VirusTotal", "type": "Multi-AV", "weight": 0.8}
    ]
    
    results = []
    for source in sources:
        verdict_options = ["Clean", "Suspicious", "Malicious", "No Data", "Establishing"]
        weights = [0.6, 0.15, 0.1, 0.1, 0.05]
        verdict = random.choices(verdict_options, weights=weights)[0]
        
        results.append({
            "source": source["name"],
            "type": source["type"],
            "verdict": verdict,
            "last_checked": datetime.now() - timedelta(hours=random.randint(0, 24)),
            "confidence": random.randint(60, 95) if verdict != "No Data" else None
        })
    
    return results

def generate_analyst_summary(verdict: str) -> Tuple[str, str]:
    """Generate AI/analyst summary"""
    summaries = {
        "Malicious": (
            "This domain exhibits characteristics consistent with malicious infrastructure. "
            "Analysis indicates active involvement in command-and-control communications "
            "associated with documented threat actor activity. Multiple independent sources "
            "confirm its role in supporting offensive cyber operations.",
            "Technical indicators suggest this domain serves as a C2 server for APT28/Sandworm "
            "activity, with connections to documented phishing campaigns targeting government "
            "entities and critical infrastructure. The domain resolves to infrastructure that has "
            "been previously associated with credential harvesting operations. SSL certificate "
            "patterns and DNS records align with established malware deployment frameworks."
        ),
        "Suspicious": (
            "This domain displays anomalous characteristics warranting additional scrutiny. "
            "While not definitively classified as malicious, several indicators deviate from "
            "expected patterns for legitimate web infrastructure. The registration and hosting "
            "characteristics suggest potential misuse but lack conclusive evidence.",
            "Detailed investigation reveals a combination of concerning factors: recent registration "
            "with privacy-protected WHOIS data, hosting on infrastructure associated with known "
            "bulletproof providers, and content patterns inconsistent with stated purpose. However, "
            "no direct malicious activity has been observed. Recommend continued monitoring and "
            "periodic reassessment as additional data becomes available."
        ),
        "Benign": (
            "This domain demonstrates characteristics consistent with legitimate web presence. "
            "Comprehensive analysis across multiple intelligence sources reveals no indicators "
            "of malicious activity. The domain is associated with established business operations "
            "and maintains positive reputation across security feeds.",
            "Extended analysis confirms the domain's legitimacy through multiple verification "
            "vectors: established business registration, valid SSL certificates from trusted CAs, "
            "consistent DNS records matching known infrastructure, and presence on multiple "
            "reputation allowlists. No connections to threat actors, malware hosting, or "
            "phishing infrastructure have been identified through any source."
        ),
        "Unknown": (
            "Insufficient information is currently available to render a definitive assessment "
            "for this domain. The limited intelligence coverage prevents confident classification. "
            "Additional data collection and monitoring are required before reaching conclusions.",
            "This domain lacks sufficient historical and technical context for thorough evaluation. "
            "Key data sources report no findings, which may indicate recent registration, low "
            "visibility, or simply gaps in intelligence coverage. Continued observation over time "
            "will enable more accurate classification as behavioral patterns emerge."
        )
    }
    
    return summaries.get(verdict, summaries["Unknown"])

def generate_evidence_list(verdict: str) -> List[Dict]:
    """Generate evidence items supporting the verdict"""
    evidence_pool = {
        "Malicious": [
            {"type": "DNS", "description": "Domain resolves to IP 185.234.72.14 - documented C2 node in threat feeds", "source": "Passive DNS", "weight": 0.95},
            {"type": "Reputation", "description": "Blocked by 3 major threat intel providers for malware distribution", "source": "Reputation Feed", "weight": 0.90},
            {"type": "Web", "description": "Historical content shows redirection to credential harvesting pages", "source": "URL Scanner", "weight": 0.88},
            {"type": "WHOIS", "description": "Registration data matches previously identified APT infrastructure pattern", "source": "WHOIS DB", "weight": 0.75},
            {"type": "SSL", "description": "Certificate issued by suspicious CA with known malicious history", "source": "Cert Transparency", "weight": 0.82}
        ],
        "Suspicious": [
            {"type": "DNS", "description": "Uses dynamic DNS provider commonly abused by threat actors", "source": "Passive DNS", "weight": 0.70},
            {"type": "WHOIS", "description": "Registration through privacy protection service", "source": "WHOIS DB", "weight": 0.60},
            {"type": "Reputation", "description": "Flagged by 1 source for suspicious activity patterns", "source": "Reputation Feed", "weight": 0.55},
            {"type": "Infrastructure", "description": "Hosted on AS with history of abuse complaints", "source": "BGP View", "weight": 0.50},
            {"type": "Web", "description": "Minimal historical web content available", "source": "Web Search", "weight": 0.40}
        ],
        "Benign": [
            {"type": "DNS", "description": "Stable DNS records matching legitimate hosting infrastructure", "source": "Passive DNS", "weight": 0.95},
            {"type": "Reputation", "description": "Clean status across all major security feeds", "source": "Reputation Feed", "weight": 0.98},
            {"type": "WHOIS", "description": "Verified business registration with valid contact information", "source": "WHOIS DB", "weight": 0.90},
            {"type": "SSL", "description": "Valid certificate from established Certificate Authority", "source": "Cert Transparency", "weight": 0.95},
            {"type": "Web", "description": "Consistent business website and active services", "source": "URL Scanner", "weight": 0.92}
        ],
        "Unknown": [
            {"type": "DNS", "description": "No historical DNS records available", "source": "Passive DNS", "weight": 0.30},
            {"type": "WHOIS", "description": "Insufficient registration data for analysis", "source": "WHOIS DB", "weight": 0.25},
            {"type": "Reputation", "description": "No reputation data available from major feeds", "source": "Reputation Feed", "weight": 0.20},
            {"type": "Web", "description": "Unable to retrieve web content for analysis", "source": "Web Search", "weight": 0.15}
        ]
    }
    
    available = evidence_pool.get(verdict, evidence_pool["Unknown"])
    num_evidence = random.randint(3, min(5, len(available)))
    selected = random.sample(available, num_evidence)
    
    for item in selected:
        item["id"] = hashlib.md5(item["description"].encode()).hexdigest()[:8]
    
    return selected

def generate_web_exposure(verdict: str) -> List[Dict]:
    """Generate web search and open source exposure results"""
    results = [
        {"title": "Security Research Report on Infrastructure", "source": "BleepingComputer", "date": "2024-01-15", "classification": "Relevant"},
        {"title": "Community Discussion - Is this domain safe?", "source": "Reddit", "date": "2024-01-10", "classification": "Informational"},
        {"title": "VirusTotal Community Analysis", "source": "VirusTotal", "date": "2024-01-08", "classification": "Relevant"},
        {"title": "Domain WHOIS Lookup Result", "source": "Whois365", "date": "2024-01-05", "classification": "Clean"},
        {"title": "Archived Website Snapshot", "source": "Wayback Machine", "date": "2023-12-20", "classification": "Informational"}
    ]
    
    if verdict == "Malicious":
        results.extend([
            {"title": "Threat Actor Profile - Domain Listed", "source": "AlienVault OTX", "date": "2024-01-12", "classification": "Suspicious"},
            {"title": "Phishing Campaign Analysis", "source": "KrebsOnSecurity", "date": "2024-01-14", "classification": "Relevant"}
        ])
    
    return results[:7]

def generate_attribution_data(domain: str) -> Dict:
    """Generate attribution and relationship data"""
    return {
        "domain": domain,
        "ip_addresses": [
            {"ip": "185.234.72.14", "asn": "AS207713", "country": "RU", "relationship": "hosts"},
            {"ip": "91.219.236.166", "asn": "AS49349", "country": "NL", "relationship": "hosts"}
        ],
        "nameservers": [
            {"ns": "ns1.cloudns.net", "relationship": "uses"},
            {"ns": "ns2.cloudns.net", "relationship": "uses"}
        ],
        "related_domains": [
            {"domain": "malware-payload.net", "relationship": "sibling", "malicious": True},
            {"domain": "phishing-target.org", "relationship": "target", "malicious": True},
            {"domain": "c2-collector.com", "relationship": "sibling", "malicious": True}
        ],
        "threat_actors": [
            {"name": "APT28", "relationship": "linked", "confidence": 0.85},
            {"name": "Sandworm", "relationship": "linked", "confidence": 0.72}
        ]
    }

def generate_technical_data(domain: str) -> Dict:
    """Generate technical enrichment data"""
    return {
        "dns": {
            "A": ["185.234.72.14", "91.219.236.166"],
            "AAAA": ["2a04:52c0:1014::14"],
            "MX": ["mail." + domain],
            "NS": ["ns1.cloudns.net", "ns2.cloudns.net", "ns3.cloudns.net"],
            "TXT": ["v=spf1 include:_spf.google.com ~all", "google-site-verification=xxx"],
            "SOA": ["ns1.cloudns.net. admin." + domain + ". 2024011501 7200 3600 1209600 86400"]
        },
        "whois": {
            "registrar": "CloudNs",
            "registrant_name": "REDACTED FOR PRIVACY",
            "registrant_org": "Private Person",
            "registrant_country": "RU",
            "creation_date": "2023-06-15",
            "expiration_date": "2025-06-15",
            "updated_date": "2024-01-10",
            "nameservers": ["ns1.cloudns.net", "ns2.cloudns.net"]
        },
        "reputation": {
            "spam_score": random.randint(0, 30),
            "phishing_score": random.randint(0, 50),
            "malware_score": random.randint(0, 70),
            "overall_score": random.randint(20, 80),
            "blocklist_count": random.randint(0, 5),
            "last_roboxed": (datetime.now() - timedelta(days=random.randint(1, 30))).strftime("%Y-%m-%d")
        }
    }

def get_verdict_color(verdict: str) -> str:
    """Get color scheme for verdict"""
    colors = {
        "Malicious": "#dc3545",
        "Suspicious": "#ffc107",
        "Benign": "#28a745",
        "Unknown": "#6c757d",
        "Clean": "#28a745",
        "No Data": "#adb5bd",
        "Establishing": "#17a2b8"
    }
    return colors.get(verdict, "#6c757d")

def get_classification_color(classification: str) -> str:
    """Get color for classification badge"""
    colors = {
        "Relevant": "#dc3545",
        "Suspicious": "#ffc107",
        "Informational": "#17a2b8",
        "Clean": "#28a745"
    }
    return colors.get(classification, "#6c757d")

def create_relationship_graph(data: Dict) -> str:
    """Create network visualization for relationships"""
    import networkx as nx
    
    G = nx.Graph()
    
    # Add center node
    center = data["domain"]
    G.add_node(center, label=center, color="#6366f1", size=30)
    
    # Add IP addresses
    for ip in data["ip_addresses"]:
        G.add_node(ip["ip"], label=ip["ip"], color="#f59e0b", size=20)
        G.add_edge(center, ip["ip"], label=ip["relationship"])
    
    # Add nameservers
    for ns in data["nameservers"]:
        G.add_node(ns["ns"], label=ns["ns"], color="#10b981", size=15)
        G.add_edge(center, ns["ns"], label=ns["relationship"])
    
    # Add related domains
    for rd in data["related_domains"]:
        color = "#ef4444" if rd["malicious"] else "#6b7280"
        G.add_node(rd["domain"], label=rd["domain"], color=color, size=18)
        G.add_edge(center, rd["domain"], label=rd["relationship"])
    
    # Add threat actors
    for ta in data["threat_actors"]:
        G.add_node(ta["name"], label=f"{ta['name']} ({ta['confidence']*100:.0f}%)", color="#ec4899", size=22)
        G.add_edge(center, ta["name"], label=ta["relationship"])
    
    return G