From ad69c6a2663549939c4855a790153f0f9feac679 Mon Sep 17 00:00:00 2001 From: bot Date: Wed, 29 Apr 2026 11:24:23 +0300 Subject: [PATCH] Ransomware.live integration, 48h window, CISA feeds, validation updates, README --- .gitignore | 6 ++ README.md | 119 ++++++++++++++++------- ransomware_fetcher.py | 179 ++++++++++++++++++++++++++++++++++ threat_intel_bot.py | 42 +++++--- validation/review.py | 123 ++++++++++++++++++++++++ validation/run_validation.py | 180 +++++++++++++++++++++++++++++++++++ 6 files changed, 604 insertions(+), 45 deletions(-) create mode 100644 ransomware_fetcher.py create mode 100644 validation/review.py create mode 100644 validation/run_validation.py diff --git a/.gitignore b/.gitignore index 2657ed5..0fd2914 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,12 @@ subscribers.json seen_articles.json seen_articles.db seen_articles.db-journal +seen_victims.db + +# Validation output +validation/results.jsonl +validation/seen_validation.db +validation/seen_validation_victims.db # Python __pycache__/ diff --git a/README.md b/README.md index 910398b..fe7c12e 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # telegram-rss-bot -Telegram bot that monitors cybersecurity RSS feeds and delivers real-time alerts. Articles are classified by severity and quality before delivery. +Telegram bot that monitors cybersecurity RSS feeds and ransomware.live victim data, delivering real-time alerts to Telegram topics. Articles are enriched with CVEs, threat actors, and malware families extracted from content. ## Project Structure @@ -8,17 +8,22 @@ Telegram bot that monitors cybersecurity RSS feeds and delivers real-time alerts rss_telegram_bot/ ├── threat_intel_bot.py # Main bot — commands, subscriptions, alert dispatch ├── rss_manager.py # Feed fetching, dedup, message formatting -├── content_classifier.py # Severity/quality scoring, CVE/actor/malware extraction +├── content_classifier.py # CVE/actor/malware extraction +├── ransomware_fetcher.py # ransomware.live PRO API — victim feed ├── check_feeds.py # CLI utility to check feed health ├── feeds/ │ ├── news_feeds.json │ ├── malware_feeds.json -│ ├── threat_intel_feeds.json +│ ├── threat_intel_feeds.json # includes CISA Advisories, Alerts, ICS │ ├── osint_feeds.json │ └── research_feeds.json -├── .env # Not committed — BOT_TOKEN + optional filters -├── subscribers.json # Auto-managed — chat/topic subscriptions -└── seen_articles.db # SQLite — tracks sent articles (7-day retention) +├── validation/ +│ ├── run_validation.py # continuous validation monitor (saves to results.jsonl) +│ └── review.py # pretty-print and filter results.jsonl +├── .env # not committed — see Setup +├── subscribers.json # auto-managed — chat/topic subscriptions +├── seen_articles.db # SQLite — tracks sent RSS articles (7-day retention) +└── seen_victims.db # SQLite — tracks sent ransomware victims (7-day retention) ``` ## Setup @@ -31,6 +36,7 @@ Create `.env`: ```env BOT_TOKEN=your_token_here +RANSOMWARE_LIVE_API_KEY=your_key_here # optional — register at api-pro.ransomware.live ``` Run: @@ -52,23 +58,9 @@ Categories: `news`, `malware`, `threat_intel`, `osint`, `research` **Telegram Topics:** Run `/on_` inside each topic to route categories to separate threads. -## Feed Management - -Each `feeds/*.json` file follows this structure: - -```json -{ - "category_name": { - "Feed Display Name": "https://example.com/rss.xml" - } -} -``` - -Add or remove feeds by editing the JSON, then restart the bot. Run `python3 check_feeds.py` to verify feed health before deploying. - ## Message Format -Each alert follows this structure: +RSS articles: ``` {emoji} Title @@ -83,19 +75,69 @@ Description 🔗 Read Full Article ``` -The enrichment lines (CVEs, actors, malware) only appear when the classifier finds something — clean articles stay minimal. +Ransomware victims (posted to the malware topic): + +``` +🦠 company.com claimed by LockBit + +Sector: Finance | Country: US +Data description from leak post + +📡 ransomware.live · Published (UTC) +🦠 Lockbit + +🔗 View on ransomware.live +``` + +Enrichment lines only appear when the classifier finds something — clean articles stay minimal. + +## Data Sources + +### RSS Feeds (64 feeds across 5 categories) + +| Category | Sources | +|---|---| +| `news` | Krebs, Bleeping Computer, The Hacker News, Security Week, Dark Reading, The Record, Cyberscoop, NCSC UK, and others | +| `threat_intel` | CISA Advisories, CISA Alerts, CISA ICS, SANS ISC, Unit42, Mandiant, Cisco Talos, Microsoft MSRC, and others | +| `malware` | Malware Traffic Analysis, Kaspersky Securelist, Security Affairs, Sekoia, and others | +| `osint` | DataBreaches.net, Have I Been Pwned, UpGuard, Vulmon Research | +| `research` | Google Project Zero, watchTowr Labs, SpecterOps, Synacktiv, Doyensec, NCC Group, and others | + +Add or remove feeds by editing the relevant `feeds/*.json` file, then restart the bot. Run `python3 check_feeds.py` to verify health before deploying. + +### ransomware.live PRO API + +When `RANSOMWARE_LIVE_API_KEY` is set, the bot polls the PRO API every 5 minutes for newly discovered ransomware victims. New entries are posted to the `malware` topic. Victims are tracked in `seen_victims.db` with 7-day retention. ## Classification `content_classifier.py` extracts the following from article title + description: -**CVEs** — standard `CVE-YYYY-NNNNN` pattern matching. +**CVEs** — `CVE-YYYY-NNNNN` regex pattern. -**Threat actors** — APT groups and named adversaries: APT28/29/40/41, Lazarus, Sandworm, Volt Typhoon, Scattered Spider, FIN7, and others defined in `THREAT_ACTORS`. +**Threat actors** — APT groups and named adversaries including APT28/29/40/41, Lazarus, Sandworm, Volt Typhoon, Scattered Spider, FIN7, and others. Full list 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 with word-boundary checking to avoid false positives. +**Malware families** — ransomware, C2 frameworks, loaders, stealers, APT tooling: LockBit, Cobalt Strike, Emotet, QakBot, Sliver, PlugX, and others. Full list in `MALWARE_FAMILIES`. Matching is case-insensitive with word-boundary checking. To add a family, append its display name to the set. -Severity and quality scoring were intentionally removed — the extracted enrichment fields (CVEs, actors, malware) give the reader enough context to judge importance themselves. +Severity and quality scoring are intentionally absent — the enrichment fields give enough context for the reader to judge importance. + +## Validation + +A separate monitor for testing without affecting the production bot: + +```bash +# runs continuously, same poll cycle as the bot +python3 validation/run_validation.py + +# review collected results +python3 validation/review.py # all results +python3 validation/review.py --category research # filter by category +python3 validation/review.py --has cves # only articles with CVEs +python3 validation/review.py --has malware # only articles with malware hits +python3 validation/review.py --today # only today's run +``` + +Results are saved to `validation/results.jsonl`. Each line is a JSON object with a `type` field (`article` or `victim`) so RSS articles and ransomware victims can be filtered separately. ## Systemd Service (VPS) @@ -108,10 +150,10 @@ After=network.target [Service] Type=simple -User=your_user -WorkingDirectory=/home/your_user/rss_telegram_bot -EnvironmentFile=/home/your_user/rss_telegram_bot/.env -ExecStart=/usr/bin/python3 threat_intel_bot.py +User=root +WorkingDirectory=/root/tele-bots/rss_splited_bot +EnvironmentFile=/root/tele-bots/rss_splited_bot/.env +ExecStart=/root/tele-bots/rss_splited_bot/venv/bin/python3 threat_intel_bot.py Restart=always RestartSec=10 @@ -125,9 +167,18 @@ sudo systemctl enable --now rss-bot sudo journalctl -u rss-bot -f ``` +## Deploying Updates + +From the local project directory: + +```bash +rsync -avz ransomware_fetcher.py threat_intel_bot.py rss_manager.py content_classifier.py feeds/threat_intel_feeds.json feeds/news_feeds.json feeds/research_feeds.json root@deployer:/root/tele-bots/rss_splited_bot/ +ssh root@deployer "systemctl restart rss-bot" +``` + ## Notes -- First run marks all current articles as seen — no flood on startup -- Polling interval: 5 minutes (`asyncio.sleep(300)` in `threat_intel_bot.py`) -- Only today's UTC articles are processed (strict date gate in `rss_manager.py`) -- `seen_articles.db` auto-purges entries older than 7 days +- First run marks all current articles and victims as seen — no flood on startup +- Polling interval: 5 minutes +- Articles published within the last 48 hours are processed (rolling window, not calendar day) +- `seen_articles.db` and `seen_victims.db` auto-purge entries older than 7 days diff --git a/ransomware_fetcher.py b/ransomware_fetcher.py new file mode 100644 index 0000000..e44c8c3 --- /dev/null +++ b/ransomware_fetcher.py @@ -0,0 +1,179 @@ +#!/usr/bin/env python3 +""" +ransomware.live PRO API fetcher. +Polls for new ransomware victims and converts them to article dicts +compatible with the existing send_alert pipeline. +""" + +import asyncio +import aiohttp +import sqlite3 +import logging +from datetime import datetime, timezone, timedelta +from typing import Dict, List, Optional, Tuple +from urllib.parse import urlparse +from html import escape + +logger = logging.getLogger(__name__) + +API_BASE = "https://api-pro.ransomware.live" + + +class RansomwareFetcher: + + def __init__(self, api_key: str, seen_file: str = "seen_victims.db"): + self.api_key = api_key + self.seen_file = seen_file + self._ensure_db() + + def _ensure_db(self): + with sqlite3.connect(self.seen_file) as conn: + conn.execute(""" + CREATE TABLE IF NOT EXISTS seen_victims ( + victim_id TEXT PRIMARY KEY, + seen_at TEXT NOT NULL + ) + """) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_seen_victims_seen_at ON seen_victims (seen_at)" + ) + conn.commit() + + def _is_seen(self, victim_id: str) -> bool: + with sqlite3.connect(self.seen_file) as conn: + return conn.execute( + "SELECT 1 FROM seen_victims WHERE victim_id = ?", (victim_id,) + ).fetchone() is not None + + def mark_seen(self, victim_id: str): + now = datetime.now(timezone.utc).isoformat() + cutoff = (datetime.now(timezone.utc) - timedelta(days=7)).isoformat() + with sqlite3.connect(self.seen_file) as conn: + conn.execute("DELETE FROM seen_victims WHERE seen_at < ?", (cutoff,)) + conn.execute( + "INSERT OR REPLACE INTO seen_victims (victim_id, seen_at) VALUES (?, ?)", + (victim_id, now) + ) + conn.commit() + + async def _fetch_month( + self, session: aiohttp.ClientSession, year: int, month: int + ) -> List[Dict]: + url = f"{API_BASE}/victims/" + params = {"year": year, "month": f"{month:02d}"} + headers = {"X-Api-Key": self.api_key} + try: + async with session.get(url, params=params, headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as resp: + if resp.status == 200: + data = await resp.json() + return data.get("victims", []) + elif resp.status == 401: + logger.error("ransomware.live: invalid API key") + elif resp.status == 429: + logger.warning("ransomware.live: rate limited, skipping this poll") + else: + logger.error(f"ransomware.live: HTTP {resp.status}") + except asyncio.TimeoutError: + logger.error("ransomware.live: request timed out") + except Exception as e: + logger.error(f"ransomware.live: {e}") + return [] + + async def fetch_new_victims( + self, session: aiohttp.ClientSession, initial_run: bool = False + ) -> List[Dict]: + now = datetime.now(timezone.utc) + cutoff = now - timedelta(hours=48) + + months = [(now.year, now.month)] + if now.day <= 2: + prev = (now.replace(day=1) - timedelta(days=1)) + months.append((prev.year, prev.month)) + + all_victims: List[Dict] = [] + for year, month in months: + victims = await self._fetch_month(session, year, month) + all_victims.extend(victims) + + new_victims = [] + for v in all_victims: + discovered_str = v.get("discovered", "") + if not discovered_str: + continue + try: + discovered_dt = datetime.fromisoformat(discovered_str.replace("Z", "+00:00")) + except ValueError: + continue + if discovered_dt < cutoff: + continue + + vid = v.get("id", "") + if not vid: + continue + if self._is_seen(vid): + continue + + if initial_run: + self.mark_seen(vid) + else: + new_victims.append(v) + + return new_victims + + @staticmethod + def _extract_company(victim: Dict) -> str: + raw = victim.get("victim", "") + if raw and not raw.startswith("http"): + return raw + domain = victim.get("website", "") + if domain: + return domain + if raw.startswith("http"): + return urlparse(raw).netloc or raw + return "Unknown" + + @staticmethod + def to_article(victim: Dict) -> Dict: + company = RansomwareFetcher._extract_company(victim) + group = victim.get("group", "unknown") + country = victim.get("country", "") + sector = victim.get("activity", "") + raw_desc = victim.get("description") or "" + + # Drop AI-generated placeholder descriptions + description = "" if "[AI generated]" in raw_desc else raw_desc[:300] + + meta_parts = [] + if sector: meta_parts.append(f"Sector: {sector}") + if country: meta_parts.append(f"Country: {country}") + full_desc = " | ".join(meta_parts) + if description: + full_desc = f"{full_desc}\n{description}" if full_desc else description + + discovered = victim.get("discovered", "") + try: + dt = datetime.fromisoformat(discovered.replace("Z", "+00:00")) + published_human = dt.strftime("%Y-%m-%d %H:%M UTC") + except (ValueError, AttributeError): + published_human = discovered[:10] if discovered else "Unknown" + + screenshot = victim.get("screenshot") or None + if screenshot and not screenshot.startswith("http"): + screenshot = None + + return { + "title": f"{company} claimed by {group.title()}", + "description": full_desc, + "url": victim.get("permalink", ""), + "published_human": published_human, + "source": "ransomware.live", + "category": "malware", + "feed_type": "malware", + "thumbnail": screenshot, + "cves": [], + "threat_actors": [], + "malware_families": [group.title()], + "mitre_techniques": [], + "iocs": {}, + "_victim_id": victim.get("id", ""), + } diff --git a/threat_intel_bot.py b/threat_intel_bot.py index cad188e..bd53daa 100644 --- a/threat_intel_bot.py +++ b/threat_intel_bot.py @@ -24,8 +24,10 @@ from telegram.ext import ( from telegram.constants import ParseMode from telegram.error import TelegramError +import aiohttp from rss_manager import RSSFeedManager from content_classifier import ContentClassifier +from ransomware_fetcher import RansomwareFetcher # Configure logging logging.basicConfig( @@ -54,6 +56,8 @@ class ThreatIntelBot: self.classifier = ContentClassifier() self.application = None self.monitoring_task = None + ransomware_key = os.getenv("RANSOMWARE_LIVE_API_KEY", "") + self.ransomware_fetcher = RansomwareFetcher(ransomware_key) if ransomware_key else None self.load_subscribers() def load_subscribers(self): @@ -393,19 +397,23 @@ class ThreatIntelBot: """Background task to monitor RSS feeds by category.""" logger.info("Starting RSS feed monitoring for category feeds...") - # First run - just mark existing articles as seen, don't send alerts + # First run - mark existing articles and victims as seen, don't send alerts try: for category, cfg in CATEGORY_CONFIG.items(): async with RSSFeedManager(feeds_file=cfg["feeds_file"], feed_type=category) as manager: logger.info(f"Initial {category.upper()} feed scan - marking existing articles as seen...") - existing_articles = await manager.fetch_all_feeds(initial_run=True) - logger.info( - f"Completed initial {category.upper()} scan. " - f"Found {len(existing_articles)} existing articles (no alerts sent)" - ) + await manager.fetch_all_feeds(initial_run=True) except Exception as e: logger.error(f"Error in initial feed scan: {e}") + if self.ransomware_fetcher: + try: + async with aiohttp.ClientSession() as session: + await self.ransomware_fetcher.fetch_new_victims(session, initial_run=True) + logger.info("Initial ransomware scan complete — existing victims marked as seen") + except Exception as e: + logger.error(f"Error in initial ransomware scan: {e}") + while True: try: all_articles = [] @@ -417,7 +425,7 @@ class ThreatIntelBot: if category_articles: logger.info(f"Found {len(category_articles)} new {category.upper()} articles") - # Deduplicate within this polling batch (cross-feed/source duplicates). + # Deduplicate within this polling batch unique_articles = [] batch_seen_url = set() batch_seen_content = set() @@ -425,16 +433,14 @@ class ThreatIntelBot: fp = RSSFeedManager.get_article_fingerprints(article) url_key = fp['url_key'] content_key = fp['content_key'] - if (url_key and url_key in batch_seen_url) or (content_key in batch_seen_content): continue - if url_key: batch_seen_url.add(url_key) batch_seen_content.add(content_key) unique_articles.append(article) - # Classify and send all unique articles + # Classify and send RSS articles sent_count = 0 for article in unique_articles: classified = self.classifier.classify_article(article) @@ -448,10 +454,24 @@ class ThreatIntelBot: else: logger.info("No new articles found") + # Poll ransomware.live for new victims + if self.ransomware_fetcher: + async with aiohttp.ClientSession() as session: + new_victims = await self.ransomware_fetcher.fetch_new_victims(session) + victim_count = 0 + for victim in new_victims: + article = RansomwareFetcher.to_article(victim) + sent = await self.send_alert(article) + if sent: + self.ransomware_fetcher.mark_seen(victim.get("id", "")) + victim_count += 1 + await asyncio.sleep(1) + if new_victims: + logger.info(f"Ransomware victims: {len(new_victims)} new, {victim_count} delivered") + except Exception as e: logger.error(f"Error in feed monitoring: {e}") - # Wait 5 minutes before next check await asyncio.sleep(300) def build_application(self): diff --git a/validation/review.py b/validation/review.py new file mode 100644 index 0000000..b391c59 --- /dev/null +++ b/validation/review.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +""" +Review validation results — pretty-prints results.jsonl with optional filters. + +Usage: + python3 validation/review.py # all results + python3 validation/review.py --category research # filter by category + python3 validation/review.py --severity critical # filter by severity + python3 validation/review.py --has cves # only articles with CVEs + python3 validation/review.py --has malware # only articles with malware hits + python3 validation/review.py --has actors # only articles with threat actors + python3 validation/review.py --today # only today's run +""" + +import json +import sys +import argparse +from pathlib import Path +from datetime import datetime, timezone + +RESULTS_FILE = Path(__file__).parent / "results.jsonl" + +SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "unknown": 4} + + +def parse_args(): + p = argparse.ArgumentParser() + p.add_argument("--category", help="Filter by category (news, malware, threat_intel, osint, research)") + p.add_argument("--severity", help="Filter by severity (critical, high, medium, low)") + p.add_argument("--has", choices=["cves", "malware", "actors"], help="Only show articles with these detections") + p.add_argument("--today", action="store_true", help="Only show results from today") + p.add_argument("--limit", type=int, default=0, help="Max articles to show (0 = all)") + return p.parse_args() + + +def load_results(args): + if not RESULTS_FILE.exists(): + print("No results.jsonl found — run run_validation.py first.") + sys.exit(0) + + today = datetime.now(timezone.utc).date().isoformat() + records = [] + + with open(RESULTS_FILE) as f: + for line in f: + line = line.strip() + if not line: + continue + try: + r = json.loads(line) + except json.JSONDecodeError: + continue + + if args.today and not r.get("run_at", "").startswith(today): + continue + if args.category and r.get("category") != args.category: + continue + if args.severity and r.get("severity") != args.severity: + continue + if args.has == "cves" and not r.get("cves"): + continue + if args.has == "malware" and not r.get("malware_families"): + continue + if args.has == "actors" and not r.get("threat_actors"): + continue + + records.append(r) + + records.sort(key=lambda x: SEVERITY_ORDER.get(x.get("severity", "unknown"), 4)) + return records + + +def print_record(r): + sev = r.get("severity", "unknown").upper() + sev_icons = {"CRITICAL": "🚨", "HIGH": "🔴", "MEDIUM": "🟠", "LOW": "🟡"} + icon = sev_icons.get(sev, "⚪") + + print(f"\n{'─' * 80}") + print(f"{icon} [{r.get('category', '?').upper()}] {r.get('title', 'No title')}") + print(f" 📡 {r.get('source')} · {r.get('published')} · Score: {r.get('quality_score', '?')}/100") + + if r.get("cves"): + print(f" 🆔 {', '.join(r['cves'])}") + if r.get("threat_actors"): + print(f" 👤 {', '.join(r['threat_actors'])}") + if r.get("malware_families"): + print(f" 🦠 {', '.join(r['malware_families'])}") + if r.get("mitre_techniques"): + print(f" 🎯 {', '.join(r['mitre_techniques'][:5])}") + if r.get("description"): + print(f" {r['description'][:200]}...") + print(f" 🔗 {r.get('url', '')}") + + +def main(): + args = parse_args() + records = load_results(args) + + if not records: + print("No records match the filters.") + return + + if args.limit: + records = records[:args.limit] + + total = len(records) + sev_counts = {} + for r in records: + s = r.get("severity", "unknown") + sev_counts[s] = sev_counts.get(s, 0) + 1 + + print(f"\n=== Validation Results ({total} articles) ===") + print(" " + " ".join(f"{s.upper()}: {c}" for s, c in sorted(sev_counts.items(), key=lambda x: SEVERITY_ORDER.get(x[0], 4)))) + + for r in records: + print_record(r) + + print(f"\n{'─' * 80}") + print(f"Total: {total} articles") + + +if __name__ == "__main__": + main() diff --git a/validation/run_validation.py b/validation/run_validation.py new file mode 100644 index 0000000..5d82ed8 --- /dev/null +++ b/validation/run_validation.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +""" +Validation script — continuously fetches RSS feeds + ransomware.live victims, +classifies articles, appends everything to results.jsonl. +Runs independently from the bot using its own seen DBs. + +Usage: + python3 validation/run_validation.py + Ctrl+C to stop +""" + +import asyncio +import aiohttp +import json +import os +import sys +from datetime import datetime, timezone +from pathlib import Path + +ROOT = Path(__file__).parent.parent +sys.path.insert(0, str(ROOT)) + +# Load .env from project root +env_file = ROOT / ".env" +if env_file.exists(): + with open(env_file) as f: + for line in f: + line = line.strip() + if line and not line.startswith("#") and "=" in line: + k, v = line.split("=", 1) + os.environ.setdefault(k.strip(), v.strip().strip("\"'")) + +from rss_manager import RSSFeedManager +from content_classifier import ContentClassifier +from ransomware_fetcher import RansomwareFetcher + +CATEGORY_CONFIG = { + "news": "feeds/news_feeds.json", + "malware": "feeds/malware_feeds.json", + "threat_intel": "feeds/threat_intel_feeds.json", + "osint": "feeds/osint_feeds.json", + "research": "feeds/research_feeds.json", +} + +OUTPUT_FILE = Path(__file__).parent / "results.jsonl" +SEEN_FILE = str(Path(__file__).parent / "seen_validation.db") +SEEN_VICTIMS_FILE = str(Path(__file__).parent / "seen_validation_victims.db") +POLL_INTERVAL = 300 + + +def save_record(f, record: dict): + f.write(json.dumps(record) + "\n") + flags = [] + if record.get("cves"): flags.append(f"CVEs: {', '.join(record['cves'][:3])}") + if record.get("threat_actors"): flags.append(f"Actors: {', '.join(record['threat_actors'][:2])}") + if record.get("malware_families"): flags.append(f"Malware: {', '.join(record['malware_families'][:2])}") + flag_str = f" → {' | '.join(flags)}" if flags else "" + category = record.get("category", "?").upper() + print(f" [{category:12s}] {record.get('title','')[:70]}{flag_str}") + + +async def poll_rss(classifier: ContentClassifier) -> int: + all_articles = [] + for category, feeds_file in CATEGORY_CONFIG.items(): + feeds_path = str(ROOT / feeds_file) + async with RSSFeedManager(feeds_file=feeds_path, seen_file=SEEN_FILE, feed_type=category) as manager: + articles = await manager.fetch_all_feeds(initial_run=False) + all_articles.extend(articles) + + if not all_articles: + return 0 + + unique = [] + seen_urls, seen_content = set(), set() + for article in all_articles: + fp = RSSFeedManager.get_article_fingerprints(article) + url_key, content_key = fp["url_key"], fp["content_key"] + if (url_key and url_key in seen_urls) or content_key in seen_content: + continue + if url_key: + seen_urls.add(url_key) + seen_content.add(content_key) + unique.append(article) + + with open(OUTPUT_FILE, "a") as f: + for article in unique: + classified = classifier.classify_article(article) + RSSFeedManager.mark_article_as_sent(classified, seen_file=SEEN_FILE) + record = { + "run_at": datetime.now(timezone.utc).isoformat(), + "type": "article", + "title": classified.get("title"), + "source": classified.get("source"), + "url": classified.get("url"), + "published": classified.get("published_human"), + "category": classified.get("category"), + "cves": classified.get("cves", []), + "threat_actors": classified.get("threat_actors", []), + "malware_families": classified.get("malware_families", []), + "mitre_techniques": classified.get("mitre_techniques", []), + "description": classified.get("description", ""), + } + save_record(f, record) + + return len(unique) + + +async def poll_victims(fetcher: RansomwareFetcher) -> int: + async with aiohttp.ClientSession() as session: + new_victims = await fetcher.fetch_new_victims(session) + + if not new_victims: + return 0 + + with open(OUTPUT_FILE, "a") as f: + for victim in new_victims: + article = RansomwareFetcher.to_article(victim) + fetcher.mark_seen(victim.get("id", "")) + record = { + "run_at": datetime.now(timezone.utc).isoformat(), + "type": "victim", + "title": article.get("title"), + "source": "ransomware.live", + "url": article.get("url"), + "published": article.get("published_human"), + "category": "malware", + "cves": [], + "threat_actors": [], + "malware_families": article.get("malware_families", []), + "mitre_techniques": [], + "description": article.get("description", ""), + } + save_record(f, record) + + return len(new_victims) + + +async def main(): + classifier = ContentClassifier() + + ransomware_key = os.getenv("RANSOMWARE_LIVE_API_KEY", "") + fetcher = RansomwareFetcher(ransomware_key, seen_file=SEEN_VICTIMS_FILE) if ransomware_key else None + + print(f"Validation monitor started — polling every {POLL_INTERVAL}s. Ctrl+C to stop.") + print(f"RSS feeds: {'enabled' if CATEGORY_CONFIG else 'disabled'}") + print(f"ransomware.live: {'enabled' if fetcher else 'disabled (no API key)'}") + print(f"Output: {OUTPUT_FILE}\n") + + # Initial scan — mark everything currently visible as seen + print(f"[{datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}] Initial scan...") + for category, feeds_file in CATEGORY_CONFIG.items(): + feeds_path = str(ROOT / feeds_file) + async with RSSFeedManager(feeds_file=feeds_path, seen_file=SEEN_FILE, feed_type=category) as manager: + await manager.fetch_all_feeds(initial_run=True) + if fetcher: + async with aiohttp.ClientSession() as session: + await fetcher.fetch_new_victims(session, initial_run=True) + print(" Done. Watching for new items from now on.\n") + + while True: + await asyncio.sleep(POLL_INTERVAL) + ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") + print(f"[{ts}] Polling...") + try: + rss_count = await poll_rss(classifier) + victim_count = await poll_victims(fetcher) if fetcher else 0 + total = rss_count + victim_count + if total: + print(f" Saved {rss_count} articles + {victim_count} victims\n") + else: + print(f" Nothing new\n") + except Exception as e: + print(f" Error: {e}\n") + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + print("\nStopped.")