From 47d8394568179a694e8f4ba2c71a996ab65e4f68 Mon Sep 17 00:00:00 2001 From: bot Date: Wed, 29 Apr 2026 09:15:53 +0300 Subject: [PATCH] Remove severity and quality scoring, classifier is extraction-only --- README.md | 18 +-- content_classifier.py | 365 ++++++------------------------------------ rss_manager.py | 25 --- threat_intel_bot.py | 39 +---- 4 files changed, 56 insertions(+), 391 deletions(-) diff --git a/README.md b/README.md index 8600168..910398b 100644 --- a/README.md +++ b/README.md @@ -31,8 +31,6 @@ Create `.env`: ```env BOT_TOKEN=your_token_here -MIN_QUALITY_SCORE=0 -ALLOWED_SEVERITIES=critical,high,medium,low ``` Run: @@ -95,21 +93,9 @@ The enrichment lines (CVEs, actors, malware) only appear when the classifier fin **Threat actors** — APT groups and named adversaries: APT28/29/40/41, Lazarus, Sandworm, Volt Typhoon, Scattered Spider, FIN7, and others defined in `THREAT_ACTORS`. -**Malware families** — ransomware, C2 frameworks, loaders, stealers, and APT tooling: LockBit, Cobalt Strike, Emotet, QakBot, Sliver, PlugX, and others defined in `MALWARE_FAMILIES`. To add a family, append its display name to the set — matching is case-insensitive. +**Malware families** — ransomware, C2 frameworks, loaders, stealers, and APT tooling: LockBit, Cobalt Strike, Emotet, QakBot, Sliver, PlugX, and others defined in `MALWARE_FAMILIES`. To add a family, append its display name to the set — matching is case-insensitive with word-boundary checking to avoid false positives. -**Severity** is keyword-based: -- `critical` — zero-days, active exploitation, RCE, ransomware -- `high` — privesc, auth bypass, code execution, kernel exploits -- `medium` — XSS, CSRF, DoS, memory corruption -- `low` — everything else - -**Quality score (0–100)** factors in: source reputation, content length, CVE presence, PoC indicators, threat actor mentions. - -Filter via env vars: -```env -MIN_QUALITY_SCORE=50 -ALLOWED_SEVERITIES=critical,high -``` +Severity and quality scoring were intentionally removed — the extracted enrichment fields (CVEs, actors, malware) give the reader enough context to judge importance themselves. ## Systemd Service (VPS) diff --git a/content_classifier.py b/content_classifier.py index 6cb568e..f70e9ac 100644 --- a/content_classifier.py +++ b/content_classifier.py @@ -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']}") diff --git a/rss_manager.py b/rss_manager.py index 37207ab..d6039ef 100644 --- a/rss_manager.py +++ b/rss_manager.py @@ -487,31 +487,6 @@ class RSSFeedManager: return message, article.get('thumbnail') - @staticmethod - def build_why_this_matters(article: Dict) -> str: - """Generate a concise operator-grade impact sentence.""" - reasons = [] - severity = str(article.get("severity", "")).lower() - cves = article.get("cves", []) or [] - actors = article.get("threat_actors", []) or [] - classes = set(article.get("classifications", []) or []) - quality = int(article.get("quality_score", 0) or 0) - - if severity in {"critical", "high"}: - reasons.append(f"{severity.upper()} severity") - if cves: - reasons.append(f"references {len(cves)} CVE(s)") - if actors: - reasons.append("mentions known threat actors") - if "poc" in classes: - reasons.append("exploit/PoC indicators present") - if "advisory" in classes: - reasons.append("actionable advisory/patch context") - if not reasons and quality >= 70: - reasons.append("high-signal technical reporting") - - return "; ".join(reasons[:3]) - # Example usage for testing async def main(): async with RSSFeedManager() as rss_manager: diff --git a/threat_intel_bot.py b/threat_intel_bot.py index 3ec694a..cad188e 100644 --- a/threat_intel_bot.py +++ b/threat_intel_bot.py @@ -47,19 +47,11 @@ class ThreatIntelBot: self, token: str, subscribers_file: str = "subscribers.json", - min_quality_score: int = 0, - allowed_severities: str = "critical,high,medium,low", ): self.token = token self.subscribers_file = subscribers_file self.subscribers: Dict[str, Dict[str, Any]] = {} self.classifier = ContentClassifier() - self.min_quality_score = max(0, min(min_quality_score, 100)) - self.allowed_severities = { - value.strip().lower() - for value in allowed_severities.split(",") - if value.strip() - } or {"critical", "high", "medium", "low"} self.application = None self.monitoring_task = None self.load_subscribers() @@ -442,29 +434,17 @@ class ThreatIntelBot: batch_seen_content.add(content_key) unique_articles.append(article) - # Send alerts for new articles + # Classify and send all unique articles sent_count = 0 - filtered_articles = [] for article in unique_articles: classified = self.classifier.classify_article(article) - if classified.get("quality_score", 0) < self.min_quality_score: - continue - if classified.get("severity", "low") not in self.allowed_severities: - continue - filtered_articles.append(classified) - - for article in filtered_articles: - sent = await self.send_alert(article) + sent = await self.send_alert(classified) if sent: sent_count += 1 - # Small delay to avoid rate limiting await asyncio.sleep(1) if unique_articles: - logger.info( - f"Processed {len(unique_articles)} unique new articles; " - f"after filters {len(filtered_articles)}; delivered {sent_count}" - ) + logger.info(f"Processed {len(unique_articles)} unique articles, delivered {sent_count}") else: logger.info("No new articles found") @@ -571,18 +551,7 @@ def main(): logger.info("2. Create .env file with: BOT_TOKEN=your_token") return - # Create and run bot - try: - min_quality_score = int(os.getenv("MIN_QUALITY_SCORE", "0")) - except ValueError: - logger.warning("Invalid MIN_QUALITY_SCORE; defaulting to 0") - min_quality_score = 0 - allowed_severities = os.getenv("ALLOWED_SEVERITIES", "critical,high,medium,low") - bot = ThreatIntelBot( - bot_token, - min_quality_score=min_quality_score, - allowed_severities=allowed_severities, - ) + bot = ThreatIntelBot(bot_token) try: asyncio.run(bot.run())