#!/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', 'conti', 'lockbit', 'blackcat', 'alphv', 'cl0p', 'clop', 'revil', 'darkside', 'nobelium', 'hafnium', 'phosphorus', 'holmium', 'strontium', 'volt typhoon', 'flax typhoon', 'mustang panda', 'winnti' } # 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_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) 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['iocs'] = iocs # Log classification logger.info( f"Classified: {article['title'][:50]}... | " f"Score: {quality_score} | Severity: {severity} | " f"Types: {', '.join(classifications)} | CVEs: {len(cves)}" ) 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']}")