Remove severity and quality scoring, classifier is extraction-only
This commit is contained in:
+50
-315
@@ -1,85 +1,19 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Content Classifier for Threat Intelligence RSS Bot
|
||||
Classifies articles by type, severity, and quality
|
||||
Extracts CVEs, threat actors, malware families, MITRE techniques, and IOCs.
|
||||
"""
|
||||
|
||||
import re
|
||||
from typing import Dict, List, Set
|
||||
from typing import Dict, List
|
||||
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',
|
||||
@@ -95,7 +29,7 @@ class ContentClassifier:
|
||||
# Ransomware
|
||||
'LockBit', 'REvil', 'BlackCat', 'ALPHV', 'Cl0p', 'Conti', 'DarkSide',
|
||||
'Ryuk', 'BlackMatter', 'Akira', 'Black Basta', 'RansomHub', 'Rhysida',
|
||||
'Medusa', 'Cactus', 'Play', 'Royal', 'Hive', 'Maze', '8Base',
|
||||
'Medusa Locker', 'Cactus Ransomware', '8Base', 'Play Ransomware',
|
||||
'Hunters International', 'Inc Ransom', 'Monti', 'Nokoyawa',
|
||||
# C2 frameworks / RATs
|
||||
'Cobalt Strike', 'Mimikatz', 'Sliver', 'Brute Ratel', 'Havoc',
|
||||
@@ -111,286 +45,87 @@ class ContentClassifier:
|
||||
'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)
|
||||
MITRE_PATTERN = re.compile(r'T\d{4}(?:\.\d{3})?', re.IGNORECASE)
|
||||
CVE_PATTERN = re.compile(r'CVE-\d{4}-\d{4,7}', re.IGNORECASE)
|
||||
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_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]))
|
||||
return list(set(cve.upper() for cve in self.CVE_PATTERN.findall(text)))
|
||||
|
||||
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]))
|
||||
return list(set(t.upper() for t in self.MITRE_PATTERN.findall(text)))
|
||||
|
||||
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))
|
||||
return list(set(
|
||||
actor.upper()
|
||||
for actor in self.THREAT_ACTORS
|
||||
if re.search(r'\b' + re.escape(actor) + r'\b', text_lower)
|
||||
))
|
||||
|
||||
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))
|
||||
return list(set(
|
||||
family
|
||||
for family in self.MALWARE_FAMILIES
|
||||
if re.search(r'\b' + re.escape(family.lower()) + r'\b', text_lower)
|
||||
))
|
||||
|
||||
def extract_iocs(self, text: str) -> Dict[str, List[str]]:
|
||||
"""Extract Indicators of Compromise from text"""
|
||||
iocs = {
|
||||
'ips': [],
|
||||
'domains': [],
|
||||
'md5': [],
|
||||
'sha1': [],
|
||||
'sha256': []
|
||||
iocs: Dict[str, List[str]] = {
|
||||
'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):
|
||||
for ip in self.IP_PATTERN.findall(text):
|
||||
if all(0 <= int(o) <= 255 for o in ip.split('.')):
|
||||
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
|
||||
iocs['md5'] = self.HASH_MD5_PATTERN.findall(text)
|
||||
iocs['sha1'] = self.HASH_SHA1_PATTERN.findall(text)
|
||||
iocs['sha256'] = self.HASH_SHA256_PATTERN.findall(text)
|
||||
iocs['domains'] = [m.group(0).lower() for m in self.DOMAIN_PATTERN.finditer(text)]
|
||||
for key in iocs:
|
||||
iocs[key] = list(set(iocs[key]))[:5] # Limit to 5 per type
|
||||
|
||||
iocs[key] = list(set(iocs[key]))[:5]
|
||||
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
|
||||
combined_text = f"{article.get('title', '')} {article.get('description', '')}"
|
||||
|
||||
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)
|
||||
cves = self.extract_cves(combined_text)
|
||||
mitre_techniques = self.extract_mitre_techniques(combined_text)
|
||||
threat_actors = self.extract_threat_actors(combined_text)
|
||||
threat_actors = self.extract_threat_actors(combined_text)
|
||||
malware_families = self.extract_malware_families(combined_text)
|
||||
iocs = self.extract_iocs(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['cves'] = cves
|
||||
article['mitre_techniques'] = mitre_techniques
|
||||
article['threat_actors'] = threat_actors
|
||||
article['threat_actors'] = threat_actors
|
||||
article['malware_families'] = malware_families
|
||||
article['iocs'] = iocs
|
||||
article['iocs'] = iocs
|
||||
|
||||
# Log classification
|
||||
logger.info(
|
||||
f"Classified: {article['title'][:50]}... | "
|
||||
f"Score: {quality_score} | Severity: {severity} | "
|
||||
f"Classified: {article.get('title', '')[:50]}... | "
|
||||
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',
|
||||
'title': 'APT28 Deploys Cobalt Strike via CVE-2024-12345 Zero-Day in Apache Struts',
|
||||
'description': 'Fancy Bear exploited a critical RCE flaw. LockBit ransomware also observed. '
|
||||
'Hash: e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855',
|
||||
'source': 'watchTowr Labs',
|
||||
'url': 'https://example.com',
|
||||
'category': 'zero_day_feeds',
|
||||
'category': 'research',
|
||||
'published': datetime.now(timezone.utc).isoformat(),
|
||||
'published_human': 'test'
|
||||
'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']}")
|
||||
c = ContentClassifier()
|
||||
result = c.classify_article(test_article)
|
||||
print(f"CVEs: {result['cves']}")
|
||||
print(f"Actors: {result['threat_actors']}")
|
||||
print(f"Malware: {result['malware_families']}")
|
||||
print(f"MITRE: {result['mitre_techniques']}")
|
||||
|
||||
Reference in New Issue
Block a user