telegram-rss-bot/content_classifier.py

#!/usr/bin/env python3
"""
Content Classifier for Threat Intelligence RSS Bot
Classifies articles by type, severity, and quality
"""

import re
from typing import Dict, List, Set
from datetime import datetime, timezone
import logging

logger = logging.getLogger(__name__)

class ContentClassifier:
    """Intelligent content classification for cybersecurity articles"""

    # High-quality research sources (get automatic quality boost)
    HIGH_QUALITY_SOURCES = {
        'watchTowr Labs', 'Doyensec Blog', 'Google Project Zero',
        'GitHub Security Lab', 'Meta Red Team X', 'Aleph Research - Posts',
        'Mandiant', 'Microsoft Security', 'Embrace The Red'
    }

    # Keywords for deep dive / technical analysis detection
    DEEP_DIVE_KEYWORDS = {
        'deep dive', 'technical analysis', 'reverse engineering', 'reversing',
        'exploitation', 'vulnerability research', 'in-depth', 'comprehensive analysis',
        'detailed analysis', 'root cause analysis', 'rca', 'technical deep-dive',
        'vulnerability analysis', 'security research', 'research paper',
        'whitepaper', 'technical report', 'forensic analysis', 'post-mortem'
    }

    # Keywords for Proof of Concept / Exploit detection
    POC_KEYWORDS = {
        'poc', 'proof of concept', 'proof-of-concept', 'exploit code',
        'exploit released', 'weaponized', 'exploit available', 'working exploit',
        'public exploit', 'exploit poc', 'exploit demo', 'demonstration',
        'exploit details', 'attack code', 'sample code', 'exploit published',
        'exploit development', 'exploit chain', 'exploit technique'
    }

    # Keywords for vulnerability/advisory detection
    VULNERABILITY_KEYWORDS = {
        'vulnerability', 'vulnerabilities', 'security flaw', 'security bug',
        'advisory', 'security advisory', 'patch', 'update', 'hotfix',
        'security update', 'bug fix', 'disclosure', 'responsible disclosure',
        'coordinated disclosure', 'full disclosure'
    }

    # Keywords for research papers/academic content
    RESEARCH_KEYWORDS = {
        'paper', 'research', 'study', 'findings', 'methodology', 'academic',
        'conference', 'presentation', 'talk', 'blackhat', 'defcon', 'rsa',
        'syscan', 'infiltrate', 'recon', 'pwn2own'
    }

    # Critical severity indicators
    CRITICAL_KEYWORDS = {
        'zero-day', '0day', 'zero day', 'actively exploited', 'active exploitation',
        'in the wild', 'itw', 'critical vulnerability', 'remote code execution', 'rce',
        'unauthenticated rce', 'pre-auth', 'wormable', 'internet-facing',
        'mass exploitation', 'ransomware', 'supply chain', 'widespread'
    }

    # High severity indicators
    HIGH_KEYWORDS = {
        'privilege escalation', 'privesc', 'authentication bypass', 'auth bypass',
        'arbitrary code execution', 'code execution', 'sandbox escape',
        'kernel exploit', 'local privilege escalation', 'lpe', 'arbitrary file write',
        'arbitrary file read', 'path traversal', 'directory traversal',
        'sql injection', 'sqli', 'command injection', 'deserialization'
    }

    # Medium severity indicators
    MEDIUM_KEYWORDS = {
        'cross-site scripting', 'xss', 'csrf', 'cross-site request forgery',
        'information disclosure', 'data leak', 'sensitive information',
        'denial of service', 'dos', 'memory corruption', 'use after free',
        'buffer overflow', 'heap overflow', 'stack overflow'
    }

    # Threat actor names (common APT groups)
    THREAT_ACTORS = {
        'apt1', 'apt28', 'apt29', 'apt32', 'apt33', 'apt34', 'apt35', 'apt37', 'apt38', 'apt39', 'apt40', 'apt41',
        'lazarus', 'kimsuky', 'andariel', 'fancy bear', 'cozy bear', 'sandworm',
        'turla', 'equation group', 'carbanak', 'fin7', 'fin6', 'fin8',
        'nobelium', 'hafnium', 'phosphorus', 'holmium', 'strontium',
        'volt typhoon', 'flax typhoon', 'mustang panda', 'winnti',
        'scattered spider', 'lapsus', 'lapsus$', 'unc2452', 'unc3944',
        'ta505', 'ta577', 'ta558', 'gold southfield', 'gold dupont',
    }

    # Stored as display names; matched case-insensitively
    MALWARE_FAMILIES = {
        # Ransomware
        'LockBit', 'REvil', 'BlackCat', 'ALPHV', 'Cl0p', 'Conti', 'DarkSide',
        'Ryuk', 'BlackMatter', 'Akira', 'Black Basta', 'RansomHub', 'Rhysida',
        'Medusa', 'Cactus', 'Play', 'Royal', 'Hive', 'Maze', '8Base',
        'Hunters International', 'Inc Ransom', 'Monti', 'Nokoyawa',
        # C2 frameworks / RATs
        'Cobalt Strike', 'Mimikatz', 'Sliver', 'Brute Ratel', 'Havoc',
        'AsyncRAT', 'Remcos', 'njRAT', 'NanoCore', 'XWorm', 'QuasarRAT',
        'DarkComet', 'NetWire', 'Metasploit',
        # Loaders / droppers
        'Emotet', 'TrickBot', 'QakBot', 'IcedID', 'BazarLoader', 'Dridex',
        'GootLoader', 'BumbleBee', 'PikaBot', 'DarkGate',
        # Stealers
        'AgentTesla', 'FormBook', 'RedLine', 'Vidar', 'Raccoon', 'Lumma',
        'Rhadamanthys', 'StealC', 'Meduza', 'Aurora',
        # APT tooling
        'PlugX', 'ShadowPad', 'Gh0stRAT', 'PoisonIvy',
    }

    # MITRE ATT&CK technique pattern
    MITRE_PATTERN = re.compile(r'T\d{4}(?:\.\d{3})?', re.IGNORECASE)

    # CVE pattern
    CVE_PATTERN = re.compile(r'CVE-\d{4}-\d{4,7}', re.IGNORECASE)

    # IOC patterns
    IP_PATTERN = re.compile(r'\b(?:\d{1,3}\.){3}\d{1,3}\b')
    DOMAIN_PATTERN = re.compile(r'\b[a-z0-9]+(?:[\-\.]{1}[a-z0-9]+)*\.[a-z]{2,6}\b', re.IGNORECASE)
    HASH_MD5_PATTERN = re.compile(r'\b[a-f0-9]{32}\b', re.IGNORECASE)
    HASH_SHA1_PATTERN = re.compile(r'\b[a-f0-9]{40}\b', re.IGNORECASE)
    HASH_SHA256_PATTERN = re.compile(r'\b[a-f0-9]{64}\b', re.IGNORECASE)

    def __init__(self):
        """Initialize classifier"""
        pass

    def extract_cves(self, text: str) -> List[str]:
        """Extract CVE identifiers from text"""
        cves = self.CVE_PATTERN.findall(text)
        return list(set([cve.upper() for cve in cves]))

    def extract_mitre_techniques(self, text: str) -> List[str]:
        """Extract MITRE ATT&CK technique IDs from text"""
        techniques = self.MITRE_PATTERN.findall(text)
        return list(set([t.upper() for t in techniques]))

    def extract_threat_actors(self, text: str) -> List[str]:
        """Extract known threat actor names from text"""
        text_lower = text.lower()
        found_actors = []
        for actor in self.THREAT_ACTORS:
            if actor in text_lower:
                found_actors.append(actor.upper())
        return list(set(found_actors))

    def extract_malware_families(self, text: str) -> List[str]:
        """Extract known malware family / tool names from text"""
        text_lower = text.lower()
        found = []
        for family in self.MALWARE_FAMILIES:
            if family.lower() in text_lower:
                found.append(family)
        return list(set(found))

    def extract_iocs(self, text: str) -> Dict[str, List[str]]:
        """Extract Indicators of Compromise from text"""
        iocs = {
            'ips': [],
            'domains': [],
            'md5': [],
            'sha1': [],
            'sha256': []
        }

        # Extract IPs (basic validation)
        potential_ips = self.IP_PATTERN.findall(text)
        for ip in potential_ips:
            octets = [int(x) for x in ip.split('.')]
            if all(0 <= x <= 255 for x in octets):
                iocs['ips'].append(ip)

        # Extract hashes
        iocs['md5'] = self.HASH_MD5_PATTERN.findall(text)
        iocs['sha1'] = self.HASH_SHA1_PATTERN.findall(text)
        iocs['sha256'] = self.HASH_SHA256_PATTERN.findall(text)

        # Extract domains
        iocs['domains'] = [match.group(0).lower() for match in self.DOMAIN_PATTERN.finditer(text)]

        # Remove duplicates
        for key in iocs:
            iocs[key] = list(set(iocs[key]))[:5]  # Limit to 5 per type

        return iocs

    def count_keywords(self, text: str, keywords: Set[str]) -> int:
        """Count keyword matches in text (case-insensitive)"""
        text_lower = text.lower()
        count = 0
        for keyword in keywords:
            if keyword in text_lower:
                count += 1
        return count

    def detect_severity(self, title: str, description: str, cves: List[str]) -> str:
        """Detect article severity level"""
        combined = f"{title} {description}".lower()

        # Critical: Zero-days, active exploitation, RCE
        if self.count_keywords(combined, self.CRITICAL_KEYWORDS) > 0:
            return 'critical'

        # High: Privilege escalation, auth bypass, code execution
        if self.count_keywords(combined, self.HIGH_KEYWORDS) > 0:
            return 'high'

        # Medium: XSS, CSRF, DoS, memory corruption
        if self.count_keywords(combined, self.MEDIUM_KEYWORDS) > 0:
            return 'medium'

        # High if multiple CVEs mentioned (likely important)
        if len(cves) >= 3:
            return 'high'

        # Default to low
        return 'low'

    def classify_content_type(self, title: str, description: str, source: str, content_length: int) -> List[str]:
        """Classify article content type"""
        combined = f"{title} {description}".lower()
        classifications = []

        # Deep Dive detection
        deep_dive_score = self.count_keywords(combined, self.DEEP_DIVE_KEYWORDS)
        if deep_dive_score >= 2 or content_length > 1000:
            classifications.append('deep_dive')

        # PoC detection
        poc_score = self.count_keywords(combined, self.POC_KEYWORDS)
        if poc_score >= 1:
            classifications.append('poc')

        # Research paper detection
        research_score = self.count_keywords(combined, self.RESEARCH_KEYWORDS)
        if research_score >= 2:
            classifications.append('research')

        # Vulnerability/Advisory detection
        vuln_score = self.count_keywords(combined, self.VULNERABILITY_KEYWORDS)
        if vuln_score >= 1:
            classifications.append('advisory')

        # Default to news if no specific classification
        if not classifications:
            classifications.append('news')

        return classifications

    def calculate_quality_score(self, article: Dict, classifications: List[str],
                                severity: str, cves: List[str], threat_actors: List[str]) -> int:
        """Calculate article quality score (0-100)"""
        score = 0

        # Base score
        score += 10

        # Source reputation boost
        if article['source'] in self.HIGH_QUALITY_SOURCES:
            score += 20

        # Content length scoring
        content_length = len(article.get('description', ''))
        if content_length > 1000:
            score += 15
        elif content_length > 500:
            score += 10
        elif content_length > 200:
            score += 5

        # Classification scoring
        if 'deep_dive' in classifications:
            score += 20
        if 'poc' in classifications:
            score += 25
        if 'research' in classifications:
            score += 15
        if 'advisory' in classifications:
            score += 10

        # Severity scoring
        severity_scores = {
            'critical': 30,
            'high': 20,
            'medium': 10,
            'low': 5
        }
        score += severity_scores.get(severity, 0)

        # CVE presence
        if cves:
            score += min(len(cves) * 5, 15)  # Max 15 points for CVEs

        # Threat actor mention
        if threat_actors:
            score += 10

        # Cap at 100
        return min(score, 100)

    def classify_article(self, article: Dict) -> Dict:
        """
        Classify an article and add classification metadata

        Returns: Article dict with added classification fields
        """
        title = article.get('title', '')
        description = article.get('description', '')
        source = article.get('source', '')
        combined_text = f"{title} {description}"

        # Extract entities
        cves = self.extract_cves(combined_text)
        mitre_techniques = self.extract_mitre_techniques(combined_text)
        threat_actors = self.extract_threat_actors(combined_text)
        malware_families = self.extract_malware_families(combined_text)
        iocs = self.extract_iocs(combined_text)

        # Classify content type
        content_length = len(description)
        classifications = self.classify_content_type(title, description, source, content_length)

        # Detect severity
        severity = self.detect_severity(title, description, cves)

        # Calculate quality score
        quality_score = self.calculate_quality_score(
            article, classifications, severity, cves, threat_actors
        )

        # Add classification data to article
        article['classifications'] = classifications
        article['severity'] = severity
        article['quality_score'] = quality_score
        article['cves'] = cves
        article['mitre_techniques'] = mitre_techniques
        article['threat_actors'] = threat_actors
        article['malware_families'] = malware_families
        article['iocs'] = iocs

        # Log classification
        logger.info(
            f"Classified: {article['title'][:50]}... | "
            f"Score: {quality_score} | Severity: {severity} | "
            f"CVEs: {len(cves)} | Actors: {len(threat_actors)} | Malware: {len(malware_families)}"
        )

        return article

    def filter_by_quality(self, articles: List[Dict], min_score: int = 0) -> List[Dict]:
        """Filter articles by minimum quality score"""
        return [a for a in articles if a.get('quality_score', 0) >= min_score]

    def filter_by_severity(self, articles: List[Dict], severities: List[str]) -> List[Dict]:
        """Filter articles by severity levels"""
        return [a for a in articles if a.get('severity') in severities]

    def filter_by_classification(self, articles: List[Dict], classifications: List[str]) -> List[Dict]:
        """Filter articles by classification types"""
        return [
            a for a in articles
            if any(c in a.get('classifications', []) for c in classifications)
        ]

# Example usage and testing
if __name__ == "__main__":
    # Test article
    test_article = {
        'title': 'Zero-Day RCE in Apache Struts: Deep Dive Technical Analysis with PoC',
        'description': 'This technical deep-dive provides a comprehensive analysis of CVE-2024-12345, '
                      'a critical zero-day remote code execution vulnerability in Apache Struts. '
                      'We present a working proof-of-concept exploit and reverse engineering of the patch. '
                      'The vulnerability was exploited by APT28 in the wild. Sample hash: '
                      'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855',
        'source': 'Google Project Zero',
        'url': 'https://example.com',
        'category': 'zero_day_feeds',
        'published': datetime.now(timezone.utc).isoformat(),
        'published_human': 'test'
    }

    classifier = ContentClassifier()
    classified = classifier.classify_article(test_article)

    print("Classification Results:")
    print(f"Title: {classified['title']}")
    print(f"Quality Score: {classified['quality_score']}/100")
    print(f"Severity: {classified['severity']}")
    print(f"Classifications: {', '.join(classified['classifications'])}")
    print(f"CVEs: {classified['cves']}")
    print(f"MITRE Techniques: {classified['mitre_techniques']}")
    print(f"Threat Actors: {classified['threat_actors']}")
    print(f"IOCs: {classified['iocs']}")