Compare commits

...

7 Commits

Author SHA1 Message Date
bot ce26202ce1 Sync README with the ransomware/dedup/retention changes, fix deploy docs
- Document the LV/EE/LT ransomware filter and 14-day retention (was
  still saying 7 days and "posts to the malware topic" after the
  ransomware split).
- Note cross-source dedup and that a restart now re-checks for missed
  articles instead of silently marking them seen.
- Setup: create a venv and install from requirements.txt instead of a
  bare pip-install line with no version pins, matching what the
  systemd unit actually expects at venv/bin/python3.
- Deploying Updates: the rsync command was missing malware_feeds.json
  and osint_feeds.json entirely, and put the feed files at the wrong
  destination path since rsync doesn't preserve the feeds/ subdirectory
  without --relative. Now includes all 5 feed files, bot_config.py,
  check_feeds.py, and requirements.txt.
- systemd block updated to the actual current deployment (ubuntu user,
  /home/ubuntu/rss-tele-bot), not the old root/tele-bots path.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-25 12:20:23 +03:00
bot 03a3c50354 Pin dependencies; drop dead severity/quality-score code from review.py
requirements.txt (new) pins python-telegram-bot, feedparser, aiohttp,
and beautifulsoup4 to their current stable versions — the repo had no
lockfile or pinned deps at all, just a pip-install line in the README.

validation/review.py still had a --severity filter and printed a
quality_score that content_classifier.py stopped producing a while
back (see "Remove severity and quality scoring, classifier is
extraction-only"). Every record scored UNKNOWN/?, so the flag could
never match anything. Removed; --category/--has/--today/--limit and
the extracted-field printing are unchanged.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-25 12:20:15 +03:00
bot e8866ce4e6 Fix restart data loss; isolate per-article failures in the poll loop
monitor_feeds() unconditionally ran the startup scan with
initial_run=True, which marks everything unseen as seen WITHOUT
alerting — not just on the bot's very first run, but on every restart.
Anything published while the service was stopped for a deploy was
silently dropped. RSSFeedManager.first_run and the new
RansomwareFetcher.is_first_run() were already available to tell the
two cases apart; monitor_feeds() now actually checks them, and does a
real fetch-and-alert pass for anything missed on a plain restart.

Also splits the per-article classify+send loop out into
_classify_and_send()/_send_ransomware_victims(), each wrapping a
single article/victim in its own try/except — previously one bad
article could raise past the whole batch and skip ransomware polling
for that cycle too. Category/poll-interval config now comes from
bot_config.py instead of a local copy.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-25 12:20:08 +03:00
bot 1b2eb74cf1 Drop unused IOC extraction
extract_iocs() (IPs, domains, hashes) ran on every classified article
but the result was never shown in Telegram alerts, never saved to
validation/results.jsonl, and never printed by review.py — pure wasted
work every poll cycle. MITRE technique extraction stays; that one is
actually used by the validation tooling.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-25 12:20:00 +03:00
bot ce3b813e8b Add cross-source duplicate detection, extend retention to 14 days
Exact URL/text matching missed the same story from a different outlet
(different URL, different wording). Add fuzzy title matching via
difflib, checked both within a single poll batch (the common case —
two feeds returning the same story in the same 5-minute cycle) and
against a rolling window of recently-sent titles in the seen-DB (the
cross-cycle case, e.g. follow-up coverage a few hours later).

seen_articles.db/seen_victims.db retention bumped from 7 to 14 days
(bot_config.SEEN_RETENTION_DAYS) — the window the new title matching
actually needs, and no reason to keep dedup history longer than that.

validation/run_validation.py: switch to the shared bot_config mapping,
and prune results.jsonl to the same 14-day window each poll instead of
growing forever — it's a testing aid, not an archive.

Also fixes a bug where a future-dated (bad clock/feed) article stayed
"recent" indefinitely instead of aging out after 48 hours, and drops
the legacy MD5 url-hash field that SHA-256 fingerprints replaced.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-25 12:19:54 +03:00
bot 1b9a5de9e3 Filter ransomware.live to LV/EE/LT, dedicated Telegram topic, harden fetcher
- New alert topic (/on_ransomware) separate from RSS malware articles,
  so a busy leak day doesn't bury other malware coverage.
- Filter victims to Latvia/Estonia/Lithuania before they ever touch the
  seen-DB — the global feed is already on ransomware.live's own site.
- Retention bumped to 14 days (bot_config.SEEN_RETENTION_DAYS).
- Add is_first_run(): lets the caller distinguish a genuinely fresh
  seen-DB from a restart of an already-running bot.
- Fix a naive-vs-aware datetime comparison that would TypeError-crash
  a poll cycle if the API ever returned a timestamp without a UTC offset.
- Drop the unused _victim_id/iocs fields from to_article()'s output.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-25 12:19:44 +03:00
bot 63f0ac74bc Add bot_config.py: single source of truth for category/feed config
Category labels, feed-file mappings, and emoji were copy-pasted across
threat_intel_bot.py, check_feeds.py, and validation/run_validation.py,
which is how the ransomware category split almost missed one of them.
check_feeds.py now imports from bot_config.py instead of keeping its
own copy.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-25 12:19:37 +03:00
10 changed files with 320 additions and 169 deletions
+24 -18
View File
@@ -7,9 +7,10 @@ Telegram bot that monitors cybersecurity RSS feeds and ransomware.live victim da
```
rss_telegram_bot/
├── threat_intel_bot.py # Main bot — commands, subscriptions, alert dispatch
├── rss_manager.py # Feed fetching, dedup, message formatting
├── content_classifier.py # CVE/actor/malware extraction
├── ransomware_fetcher.py # ransomware.live PRO API — victim feed
├── rss_manager.py # Feed fetching, dedup (incl. cross-source), message formatting
├── content_classifier.py # CVE/actor/malware/MITRE extraction
├── ransomware_fetcher.py # ransomware.live PRO API — victim feed, LV/EE/LT only
├── bot_config.py # Shared category, feed-file, and tuning config
├── check_feeds.py # CLI utility to check feed health
├── feeds/
│ ├── news_feeds.json
@@ -22,14 +23,16 @@ rss_telegram_bot/
│ └── 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)
├── seen_articles.db # SQLite — tracks sent RSS articles + titles (14-day retention)
└── seen_victims.db # SQLite — tracks sent ransomware victims (14-day retention)
```
## Setup
```bash
pip install python-telegram-bot feedparser aiohttp beautifulsoup4
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
```
Create `.env`:
@@ -54,7 +57,7 @@ python3 threat_intel_bot.py
| `/stats` | Check feed health and subscriber count |
| `/help` | Show commands |
Categories: `news`, `malware`, `threat_intel`, `osint`, `research`
Categories: `news`, `malware`, `threat_intel`, `osint`, `research`, `ransomware`
**Telegram Topics:** Run `/on_<category>` inside each topic to route categories to separate threads.
@@ -75,10 +78,10 @@ Description
🔗 Read Full Article
```
Ransomware victims (posted to the malware topic):
Ransomware victims (posted to the `ransomware` topic, separate from `malware` so a busy leak day doesn't bury other malware sources):
```
🦠 company.com claimed by LockBit
💰 company.com claimed by LockBit
Sector: Finance | Country: US
Data description from leak post
@@ -107,7 +110,7 @@ Add or remove feeds by editing the relevant `feeds/*.json` file, then restart th
### 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.
When `RANSOMWARE_LIVE_API_KEY` is set, the bot polls the PRO API every 5 minutes for newly discovered ransomware victims, filtered to companies in Latvia, Estonia, and Lithuania (`ALLOWED_RANSOMWARE_COUNTRIES` in `bot_config.py`) — the global feed is available directly on ransomware.live. New entries are posted to the `ransomware` topic (subscribe with `/on_ransomware`), kept separate from RSS `malware` articles. Victims are tracked in `seen_victims.db` with 14-day retention.
## Classification
@@ -150,10 +153,10 @@ After=network.target
[Service]
Type=simple
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
User=ubuntu
WorkingDirectory=/home/ubuntu/rss-tele-bot
EnvironmentFile=/home/ubuntu/rss-tele-bot/.env
ExecStart=/home/ubuntu/rss-tele-bot/venv/bin/python3 threat_intel_bot.py
Restart=always
RestartSec=10
@@ -167,18 +170,21 @@ sudo systemctl enable --now rss-bot
sudo journalctl -u rss-bot -f
```
Currently deployed and running this way — `enabled` (survives reboot) and `Restart=always` (survives crashes), so it no longer depends on an active SSH session.
## 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"
rsync -avz --relative ransomware_fetcher.py threat_intel_bot.py rss_manager.py content_classifier.py bot_config.py check_feeds.py requirements.txt ./feeds/threat_intel_feeds.json ./feeds/news_feeds.json ./feeds/research_feeds.json ./feeds/malware_feeds.json ./feeds/osint_feeds.json ubuntu@<vps-ip>:/home/ubuntu/rss-tele-bot/
ssh ubuntu@<vps-ip> "sudo systemctl restart rss-bot"
```
## Notes
- First run marks all current articles and victims as seen — no flood on startup
- First run (empty seen-DB) marks all current articles and victims as seen — no flood on startup. A restart of an already-initialized bot instead fetches and alerts on anything published while it was stopped, so a redeploy doesn't silently drop real alerts.
- 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
- Cross-source duplicates (the same story from a different outlet) are caught by fuzzy title matching, not just exact URL/text matches
- `seen_articles.db` and `seen_victims.db` auto-purge entries older than 14 days
+52
View File
@@ -0,0 +1,52 @@
#!/usr/bin/env python3
"""
Shared category, feed-file, and tuning configuration.
Single source of truth for values that used to be copy-pasted across
threat_intel_bot.py, check_feeds.py, and validation/run_validation.py —
edit here, not in the callers.
"""
# category -> (display label, emoji)
CATEGORY_LABELS = {
"news": ("News", "📰"),
"malware": ("Malware", "🦠"),
"threat_intel": ("Threat Intel", "🛰️"),
"osint": ("OSINT", "🕵️"),
"research": ("Research", "🔬"),
}
# category -> RSS feed list file. Every key here must also be in CATEGORY_LABELS.
CATEGORY_FEEDS = {
"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",
}
# Categories with no RSS feed of their own (e.g. fed by an API poller instead).
# Subscribable via /on_<category>, but excluded from the RSS polling/status loops.
EXTRA_CATEGORY_LABELS = {
"ransomware": ("Ransomware", "💰"),
}
CATEGORY_EMOJIS = {
key: emoji for key, (_label, emoji) in {**CATEGORY_LABELS, **EXTRA_CATEGORY_LABELS}.items()
}
# How often the bot/validation monitor polls feeds and the ransomware.live API.
POLL_INTERVAL_SECONDS = 300
# How long "seen" fingerprints and recently-sent titles are kept for dedup.
# Only used to suppress re-alerting on the same story — no other retention need.
SEEN_RETENTION_DAYS = 14
# Only alert on ransomware.live victims headquartered in these ISO-2 countries.
# The general global feed is available directly on ransomware.live's own site.
ALLOWED_RANSOMWARE_COUNTRIES = {"LV", "EE", "LT"}
# Cross-source near-duplicate title matching (catches the same story reported
# by two different outlets with different URLs/wording).
TITLE_DEDUP_WINDOW_HOURS = 72
TITLE_SIMILARITY_THRESHOLD = 0.72
+1 -8
View File
@@ -4,14 +4,7 @@ Quick script to check all RSS feeds status
"""
import asyncio
from rss_manager import RSSFeedManager
CATEGORY_FEEDS = {
"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",
}
from bot_config import CATEGORY_FEEDS
async def check_all_feeds():
totals = {}
+1 -23
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
"""
Content Classifier for Threat Intelligence RSS Bot
Extracts CVEs, threat actors, malware families, MITRE techniques, and IOCs.
Extracts CVEs, threat actors, malware families, and MITRE technique IDs.
"""
import re
@@ -47,11 +47,6 @@ class ContentClassifier:
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_SHA256_PATTERN = re.compile(r'\b[a-f0-9]{64}\b', re.IGNORECASE)
def extract_cves(self, text: str) -> List[str]:
return list(set(cve.upper() for cve in self.CVE_PATTERN.findall(text)))
@@ -75,21 +70,6 @@ class ContentClassifier:
if re.search(r'\b' + re.escape(family.lower()) + r'\b', text_lower)
))
def extract_iocs(self, text: str) -> Dict[str, List[str]]:
iocs: Dict[str, List[str]] = {
'ips': [], 'domains': [], 'md5': [], 'sha1': [], 'sha256': []
}
for ip in self.IP_PATTERN.findall(text):
if all(0 <= int(o) <= 255 for o in ip.split('.')):
iocs['ips'].append(ip)
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]
return iocs
def classify_article(self, article: Dict) -> Dict:
combined_text = f"{article.get('title', '')} {article.get('description', '')}"
@@ -97,13 +77,11 @@ class ContentClassifier:
mitre_techniques = self.extract_mitre_techniques(combined_text)
threat_actors = self.extract_threat_actors(combined_text)
malware_families = self.extract_malware_families(combined_text)
iocs = self.extract_iocs(combined_text)
article['cves'] = cves
article['mitre_techniques'] = mitre_techniques
article['threat_actors'] = threat_actors
article['malware_families'] = malware_families
article['iocs'] = iocs
logger.info(
f"Classified: {article.get('title', '')[:50]}... | "
+20 -5
View File
@@ -14,6 +14,8 @@ from typing import Dict, List, Optional, Tuple
from urllib.parse import urlparse
from html import escape
from bot_config import SEEN_RETENTION_DAYS, ALLOWED_RANSOMWARE_COUNTRIES
logger = logging.getLogger(__name__)
API_BASE = "https://api-pro.ransomware.live"
@@ -45,9 +47,15 @@ class RansomwareFetcher:
"SELECT 1 FROM seen_victims WHERE victim_id = ?", (victim_id,)
).fetchone() is not None
def is_first_run(self) -> bool:
"""True if no victims have ever been recorded as seen (fresh DB)."""
with sqlite3.connect(self.seen_file) as conn:
count = conn.execute("SELECT COUNT(1) FROM seen_victims").fetchone()[0]
return count == 0
def mark_seen(self, victim_id: str):
now = datetime.now(timezone.utc).isoformat()
cutoff = (datetime.now(timezone.utc) - timedelta(days=7)).isoformat()
cutoff = (datetime.now(timezone.utc) - timedelta(days=SEEN_RETENTION_DAYS)).isoformat()
with sqlite3.connect(self.seen_file) as conn:
conn.execute("DELETE FROM seen_victims WHERE seen_at < ?", (cutoff,))
conn.execute(
@@ -95,6 +103,13 @@ class RansomwareFetcher:
victims = await self._fetch_month(session, year, month)
all_victims.extend(victims)
# Only care about victims in these countries — the general global feed
# is available directly on ransomware.live's own site.
all_victims = [
v for v in all_victims
if (v.get("country") or "").strip().upper() in ALLOWED_RANSOMWARE_COUNTRIES
]
new_victims = []
for v in all_victims:
discovered_str = v.get("discovered", "")
@@ -104,6 +119,8 @@ class RansomwareFetcher:
discovered_dt = datetime.fromisoformat(discovered_str.replace("Z", "+00:00"))
except ValueError:
continue
if discovered_dt.tzinfo is None:
discovered_dt = discovered_dt.replace(tzinfo=timezone.utc)
if discovered_dt < cutoff:
continue
@@ -167,13 +184,11 @@ class RansomwareFetcher:
"url": victim.get("permalink", ""),
"published_human": published_human,
"source": "ransomware.live",
"category": "malware",
"feed_type": "malware",
"category": "ransomware",
"feed_type": "ransomware",
"thumbnail": screenshot,
"cves": [],
"threat_actors": [],
"malware_families": [group.title()],
"mitre_techniques": [],
"iocs": {},
"_victim_id": victim.get("id", ""),
}
+5
View File
@@ -0,0 +1,5 @@
# Hand-curated dependency pins; this project does not have a lockfile.
python-telegram-bot==22.8
feedparser==6.0.14
aiohttp==3.14.3
beautifulsoup4==4.15.0
+69 -24
View File
@@ -10,6 +10,7 @@ import logging
import asyncio
import aiohttp
import sqlite3
import difflib
from datetime import datetime, timezone, timedelta
import hashlib
from typing import Dict, List, Optional, Tuple
@@ -18,6 +19,8 @@ from html import escape
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
from pathlib import Path
from bot_config import CATEGORY_EMOJIS, SEEN_RETENTION_DAYS, TITLE_DEDUP_WINDOW_HOURS, TITLE_SIMILARITY_THRESHOLD
# Configure logging
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
@@ -32,6 +35,7 @@ class RSSFeedManager:
self.feed_type = feed_type # 'daily', 'blogs', or 'general'
self.feeds = {}
self.seen_articles = {} # Changed to dict to store timestamps
self.recent_titles: List[str] = [] # normalized titles sent recently, for cross-source dedup
self.session = None
self.first_run = True
self.load_feeds()
@@ -71,6 +75,7 @@ class RSSFeedManager:
self._ensure_seen_db()
self._migrate_legacy_seen_json()
self._load_seen_from_sqlite()
self._load_recent_titles()
def _ensure_seen_db(self):
"""Create seen-article database and indexes if missing."""
@@ -87,10 +92,34 @@ class RSSFeedManager:
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_seen_articles_seen_at ON seen_articles (seen_at)"
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS sent_titles (
title_norm TEXT NOT NULL,
seen_at TEXT NOT NULL
)
"""
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_sent_titles_seen_at ON sent_titles (seen_at)"
)
conn.commit()
except Exception as e:
logger.error(f"Error initializing seen-article database: {e}")
def _load_recent_titles(self):
"""Load normalized titles sent within the cross-source dedup window."""
try:
cutoff_iso = (datetime.now(timezone.utc) - timedelta(hours=TITLE_DEDUP_WINDOW_HOURS)).isoformat()
with sqlite3.connect(self.seen_file) as conn:
cursor = conn.execute(
"SELECT title_norm FROM sent_titles WHERE seen_at >= ?", (cutoff_iso,)
)
self.recent_titles = [row[0] for row in cursor.fetchall()]
except Exception as e:
logger.error(f"Error loading recent titles: {e}")
self.recent_titles = []
def _load_seen_from_sqlite(self):
"""Load seen keys from SQLite into memory."""
try:
@@ -149,7 +178,7 @@ class RSSFeedManager:
def save_seen_articles(self):
"""Save seen article keys with retention in SQLite."""
try:
cutoff_time = datetime.now(timezone.utc) - timedelta(days=7)
cutoff_time = datetime.now(timezone.utc) - timedelta(days=SEEN_RETENTION_DAYS)
cutoff_iso = cutoff_time.isoformat()
cleaned_articles = {}
with sqlite3.connect(self.seen_file) as conn:
@@ -166,13 +195,23 @@ class RSSFeedManager:
logger.info(f"Saved {len(self.seen_articles)} seen articles (cleaned old entries)")
except Exception as e:
logger.error(f"Error saving seen articles: {e}")
def generate_article_hash(self, article: Dict) -> str:
"""Generate unique hash for article based on URL only for better duplicate detection"""
# Use only the URL for hash to catch duplicates across different RSS sources
# Different sources may have same article with different titles/dates
url = article.get('link', '')
return hashlib.md5(url.encode()).hexdigest()
def _save_sent_title(self, title: str, seen_at: str):
"""Record a sent article's normalized title for cross-source dedup, pruning old rows."""
title_norm = self.normalize_text(title)
if not title_norm:
return
try:
cutoff_iso = (datetime.now(timezone.utc) - timedelta(days=SEEN_RETENTION_DAYS)).isoformat()
with sqlite3.connect(self.seen_file) as conn:
conn.execute("DELETE FROM sent_titles WHERE seen_at < ?", (cutoff_iso,))
conn.execute(
"INSERT INTO sent_titles (title_norm, seen_at) VALUES (?, ?)",
(title_norm, seen_at),
)
conn.commit()
except Exception as e:
logger.error(f"Error saving sent title: {e}")
@staticmethod
def canonicalize_url(url: str) -> str:
@@ -220,22 +259,36 @@ class RSSFeedManager:
}
def is_article_recent(self, published_dt: datetime) -> bool:
"""Allow articles published within the last 48 hours."""
"""Allow articles published within the last 48 hours (with brief clock-skew tolerance)."""
if published_dt.tzinfo is None:
published_dt = published_dt.replace(tzinfo=timezone.utc)
age = datetime.now(timezone.utc) - published_dt.astimezone(timezone.utc)
return age.total_seconds() <= 48 * 3600
age_seconds = (datetime.now(timezone.utc) - published_dt.astimezone(timezone.utc)).total_seconds()
return -300 <= age_seconds <= 48 * 3600
@staticmethod
def titles_are_near_duplicate(title_norm_a: str, title_norm_b: str) -> bool:
"""Fuzzy title match, tolerant of different outlets' wording for the same story."""
return difflib.SequenceMatcher(None, title_norm_a, title_norm_b).ratio() >= TITLE_SIMILARITY_THRESHOLD
def is_near_duplicate_title(self, title: str) -> bool:
"""Fuzzy-match against recently sent titles to catch the same story from a different source."""
title_norm = self.normalize_text(title)
if not title_norm:
return False
return any(self.titles_are_near_duplicate(title_norm, other) for other in self.recent_titles)
def is_duplicate_article(self, article: Dict) -> bool:
"""Check URL/content fingerprint duplicates against sent-history."""
"""Check URL/content fingerprint duplicates, plus fuzzy title matches against recent cross-source sends."""
fingerprints = self.get_article_fingerprints(article)
url_key = fingerprints['url_key']
content_key = fingerprints['content_key']
return (url_key and url_key in self.seen_articles) or (content_key in self.seen_articles)
if (url_key and url_key in self.seen_articles) or (content_key in self.seen_articles):
return True
return self.is_near_duplicate_title(article.get('title', ''))
@classmethod
def mark_article_as_sent(cls, article: Dict, seen_file: str = "seen_articles.db"):
"""Persist fingerprints only after successful delivery."""
"""Persist fingerprints and title only after successful delivery."""
manager = cls(feeds_file="", seen_file=seen_file, feed_type=article.get('feed_type', 'general'))
fingerprints = cls.get_article_fingerprints(article)
now_iso = datetime.now(timezone.utc).isoformat()
@@ -244,6 +297,7 @@ class RSSFeedManager:
manager.seen_articles[fingerprints['url_key']] = now_iso
manager.seen_articles[fingerprints['content_key']] = now_iso
manager.save_seen_articles()
manager._save_sent_title(article.get('title', ''), now_iso)
def clean_html(self, text: str) -> str:
"""Clean HTML tags and decode entities"""
@@ -322,7 +376,6 @@ class RSSFeedManager:
content = entry.description
article = {
'hash': self.generate_article_hash(entry),
'title': self.clean_html(getattr(entry, 'title', 'No Title')),
'description': self.clean_html(content),
'url': self.canonicalize_url(getattr(entry, 'link', '')),
@@ -449,15 +502,7 @@ class RSSFeedManager:
@staticmethod
def format_telegram_message(article: Dict) -> Tuple[str, Optional[str]]:
"""Format article for Telegram message"""
category_emojis = {
'news': '📰',
'malware': '🦠',
'threat_intel': '🛰️',
'osint': '🕵️',
'research': '🔬'
}
emoji = category_emojis.get(article.get('category', ''), '📰')
emoji = CATEGORY_EMOJIS.get(article.get('category', ''), '📰')
title = escape(article.get('title', 'No Title'))
description = escape(article.get('description', ''))
source = escape(article.get('source', 'Unknown Source'))
+112 -59
View File
@@ -9,7 +9,7 @@ import json
import logging
import os
from datetime import datetime, timezone
from typing import Any, Dict
from typing import Any, Dict, List
from pathlib import Path
from functools import partial
@@ -28,6 +28,7 @@ import aiohttp
from rss_manager import RSSFeedManager
from content_classifier import ContentClassifier
from ransomware_fetcher import RansomwareFetcher
from bot_config import CATEGORY_LABELS, CATEGORY_FEEDS, EXTRA_CATEGORY_LABELS, POLL_INTERVAL_SECONDS
# Configure logging
logging.basicConfig(
@@ -37,13 +38,20 @@ logging.basicConfig(
logger = logging.getLogger(__name__)
CATEGORY_CONFIG = {
"news": {"label": "News", "emoji": "📰", "feeds_file": "feeds/news_feeds.json"},
"malware": {"label": "Malware", "emoji": "🦠", "feeds_file": "feeds/malware_feeds.json"},
"threat_intel": {"label": "Threat Intel", "emoji": "🛰️", "feeds_file": "feeds/threat_intel_feeds.json"},
"osint": {"label": "OSINT", "emoji": "🕵️", "feeds_file": "feeds/osint_feeds.json"},
"research": {"label": "Research", "emoji": "🔬", "feeds_file": "feeds/research_feeds.json"},
key: {"label": label, "emoji": emoji, "feeds_file": CATEGORY_FEEDS[key]}
for key, (label, emoji) in CATEGORY_LABELS.items()
}
# Categories that aren't RSS-fed (no feeds_file) and so are excluded from the
# RSS polling/status loops in monitor_feeds() and stats_command(), but are
# still subscribable via /on_<category> and shown in /start help.
EXTRA_CATEGORIES = {
key: {"label": label, "emoji": emoji}
for key, (label, emoji) in EXTRA_CATEGORY_LABELS.items()
}
ALL_CATEGORIES = {**CATEGORY_CONFIG, **EXTRA_CATEGORIES}
class ThreatIntelBot:
def __init__(
self,
@@ -85,7 +93,7 @@ class ThreatIntelBot:
if isinstance(raw_feed_types, list):
mapped = []
for feed in raw_feed_types:
if feed in CATEGORY_CONFIG:
if feed in ALL_CATEGORIES:
mapped.append(feed)
elif feed == "daily":
mapped.extend(["news", "threat_intel", "osint", "malware"])
@@ -123,7 +131,7 @@ class ThreatIntelBot:
user_name = update.effective_user.first_name or "User"
command_lines = []
for key, cfg in CATEGORY_CONFIG.items():
for key, cfg in ALL_CATEGORIES.items():
command_lines.append(
f"{cfg['emoji']} `{('/on_' + key)}` / `{('/off_' + key)}` - {cfg['label']}"
)
@@ -151,14 +159,14 @@ class ThreatIntelBot:
async def category_on_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE, category: str):
"""Handle /on_<category> command."""
if category not in CATEGORY_CONFIG:
if category not in ALL_CATEGORIES:
return
chat_id = update.effective_chat.id
message_thread_id = getattr(update.message, 'message_thread_id', None)
subscriber_key = f"{chat_id}_{message_thread_id}" if message_thread_id else str(chat_id)
is_forum_topic = message_thread_id is not None
cfg = CATEGORY_CONFIG[category]
cfg = ALL_CATEGORIES[category]
if subscriber_key not in self.subscribers:
self.subscribers[subscriber_key] = {'topic_id': message_thread_id, 'feed_types': []}
@@ -184,14 +192,14 @@ class ThreatIntelBot:
async def category_off_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE, category: str):
"""Handle /off_<category> command."""
if category not in CATEGORY_CONFIG:
if category not in ALL_CATEGORIES:
return
chat_id = update.effective_chat.id
message_thread_id = getattr(update.message, 'message_thread_id', None)
subscriber_key = f"{chat_id}_{message_thread_id}" if message_thread_id else str(chat_id)
is_forum_topic = message_thread_id is not None
cfg = CATEGORY_CONFIG[category]
cfg = ALL_CATEGORIES[category]
if subscriber_key in self.subscribers and category in self.subscribers[subscriber_key].get('feed_types', []):
self.subscribers[subscriber_key]['feed_types'].remove(category)
@@ -224,7 +232,7 @@ class ThreatIntelBot:
if subscriber_key not in self.subscribers:
await update.message.reply_text(
"This topic is not subscribed yet. Use /on_news, /on_threat_intel, /on_malware, /on_osint, or /on_research first.",
"This topic is not subscribed yet. Use /on_news, /on_threat_intel, /on_malware, /on_osint, /on_research, or /on_ransomware first.",
message_thread_id=message_thread_id if message_thread_id else None
)
return
@@ -393,26 +401,101 @@ class ThreatIntelBot:
logger.error(f"Error sending alert: {e}")
return False
async def _classify_and_send(self, articles: List[Dict]) -> int:
"""Deduplicate a batch (exact + fuzzy cross-source title match), classify, and send.
A single bad article is logged and skipped rather than aborting the rest
of the batch (and, since ransomware polling runs after this in the caller,
rather than aborting that too).
"""
unique_articles = []
batch_seen_url = set()
batch_seen_content = set()
batch_titles: List[str] = []
for article in articles:
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
title_norm = RSSFeedManager.normalize_text(article.get('title', ''))
if title_norm and any(
RSSFeedManager.titles_are_near_duplicate(title_norm, other) for other in batch_titles
):
continue
if url_key:
batch_seen_url.add(url_key)
batch_seen_content.add(content_key)
if title_norm:
batch_titles.append(title_norm)
unique_articles.append(article)
sent_count = 0
for article in unique_articles:
try:
classified = self.classifier.classify_article(article)
sent = await self.send_alert(classified)
if sent:
sent_count += 1
except Exception as e:
logger.error(f"Error processing article {article.get('title', '')[:50]!r}: {e}")
await asyncio.sleep(1)
return sent_count
async def _send_ransomware_victims(self, victims: List[Dict]) -> int:
"""Convert and send ransomware.live victims; one bad record doesn't stop the rest."""
victim_count = 0
for victim in victims:
try:
article = RansomwareFetcher.to_article(victim)
sent = await self.send_alert(article)
if sent:
self.ransomware_fetcher.mark_seen(victim.get("id", ""))
victim_count += 1
except Exception as e:
logger.error(f"Error processing ransomware victim {victim.get('id', '')}: {e}")
await asyncio.sleep(1)
return victim_count
async def monitor_feeds(self):
"""Background task to monitor RSS feeds by category."""
logger.info("Starting RSS feed monitoring for category feeds...")
# First run - mark existing articles and victims as seen, don't send alerts
try:
for category, cfg in CATEGORY_CONFIG.items():
# Only mark-without-alerting on a genuinely fresh seen-DB. On a restart of an
# already-running bot, do a real fetch instead so articles published while the
# service was stopped (e.g. during a deploy) still get delivered.
startup_articles: List[Dict] = []
for category, cfg in CATEGORY_CONFIG.items():
try:
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...")
await manager.fetch_all_feeds(initial_run=True)
except Exception as e:
logger.error(f"Error in initial feed scan: {e}")
if manager.first_run:
logger.info(f"Initial {category.upper()} feed scan - marking existing articles as seen...")
await manager.fetch_all_feeds(initial_run=True)
else:
articles = await manager.fetch_all_feeds(initial_run=False)
if articles:
logger.info(f"Restart {category.upper()} scan - {len(articles)} article(s) missed while stopped")
startup_articles.extend(articles)
except Exception as e:
logger.error(f"Error in startup {category.upper()} scan: {e}")
if startup_articles:
sent = await self._classify_and_send(startup_articles)
logger.info(f"Delivered {sent} article(s) missed during downtime")
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")
if self.ransomware_fetcher.is_first_run():
await self.ransomware_fetcher.fetch_new_victims(session, initial_run=True)
logger.info("Initial ransomware scan complete — existing victims marked as seen")
else:
missed_victims = await self.ransomware_fetcher.fetch_new_victims(session)
if missed_victims:
logger.info(f"Restart ransomware scan - {len(missed_victims)} victim(s) missed while stopped")
await self._send_ransomware_victims(missed_victims)
except Exception as e:
logger.error(f"Error in initial ransomware scan: {e}")
logger.error(f"Error in startup ransomware scan: {e}")
while True:
try:
@@ -425,32 +508,9 @@ class ThreatIntelBot:
if category_articles:
logger.info(f"Found {len(category_articles)} new {category.upper()} articles")
# Deduplicate within this polling batch
unique_articles = []
batch_seen_url = set()
batch_seen_content = set()
for article in all_articles:
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 RSS articles
sent_count = 0
for article in unique_articles:
classified = self.classifier.classify_article(article)
sent = await self.send_alert(classified)
if sent:
sent_count += 1
await asyncio.sleep(1)
if unique_articles:
logger.info(f"Processed {len(unique_articles)} unique articles, delivered {sent_count}")
sent_count = await self._classify_and_send(all_articles)
if all_articles:
logger.info(f"Processed {len(all_articles)} candidate articles, delivered {sent_count}")
else:
logger.info("No new articles found")
@@ -458,21 +518,14 @@ class ThreatIntelBot:
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:
victim_count = await self._send_ransomware_victims(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}")
await asyncio.sleep(300)
await asyncio.sleep(POLL_INTERVAL_SECONDS)
def build_application(self):
"""Build the Telegram application"""
@@ -482,7 +535,7 @@ class ThreatIntelBot:
# Add command handlers
self.application.add_handler(CommandHandler("start", self.start_command))
self.application.add_handler(CommandHandler("help", self.help_command))
for category in CATEGORY_CONFIG:
for category in ALL_CATEGORIES:
self.application.add_handler(
CommandHandler(f"on_{category}", partial(self.category_on_command, category=category))
)
+2 -19
View File
@@ -5,7 +5,6 @@ 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
@@ -20,13 +19,10 @@ 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)")
@@ -55,8 +51,6 @@ def load_results(args):
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"):
@@ -66,18 +60,13 @@ def load_results(args):
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")
print(f"[{r.get('category', '?').upper()}] {r.get('title', 'No title')}")
print(f" 📡 {r.get('source')} · {r.get('published')}")
if r.get("cves"):
print(f" 🆔 {', '.join(r['cves'])}")
@@ -104,13 +93,7 @@ def main():
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)
+34 -13
View File
@@ -14,7 +14,7 @@ import aiohttp
import json
import os
import sys
from datetime import datetime, timezone
from datetime import datetime, timezone, timedelta
from pathlib import Path
ROOT = Path(__file__).parent.parent
@@ -33,19 +33,11 @@ if env_file.exists():
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",
}
from bot_config import CATEGORY_FEEDS as CATEGORY_CONFIG, POLL_INTERVAL_SECONDS as POLL_INTERVAL, SEEN_RETENTION_DAYS
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):
@@ -123,7 +115,7 @@ async def poll_victims(fetcher: RansomwareFetcher) -> int:
"source": "ransomware.live",
"url": article.get("url"),
"published": article.get("published_human"),
"category": "malware",
"category": article.get("category"),
"cves": [],
"threat_actors": [],
"malware_families": article.get("malware_families", []),
@@ -135,6 +127,33 @@ async def poll_victims(fetcher: RansomwareFetcher) -> int:
return len(new_victims)
def prune_results_file():
"""Keep results.jsonl bounded to the same window dedup actually needs — this is a
testing aid, not an archive, so nothing here needs to outlive SEEN_RETENTION_DAYS."""
if not OUTPUT_FILE.exists():
return
cutoff = (datetime.now(timezone.utc) - timedelta(days=SEEN_RETENTION_DAYS)).isoformat()
kept = []
dropped = 0
with open(OUTPUT_FILE) as f:
for line in f:
line = line.strip()
if not line:
continue
try:
record = json.loads(line)
except json.JSONDecodeError:
continue
if record.get("run_at", "") >= cutoff:
kept.append(line)
else:
dropped += 1
if dropped:
with open(OUTPUT_FILE, "w") as f:
f.write("\n".join(kept) + ("\n" if kept else ""))
print(f" Pruned {dropped} record(s) older than {SEEN_RETENTION_DAYS} days")
async def main():
classifier = ContentClassifier()
@@ -166,9 +185,11 @@ async def main():
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")
print(f" Saved {rss_count} articles + {victim_count} victims")
else:
print(f" Nothing new\n")
print(f" Nothing new")
prune_results_file()
print()
except Exception as e:
print(f" Error: {e}\n")