bot
364 lines
14 KiB
Python
364 lines
14 KiB
Python
#!/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']}")
|