Create a Cyber Threat Intelligence Bot

Real-time security news & ransomware alerts in Discord & Telegram

Introduction

If you’ve been in the cybersecurity community for a while, chances are you’ve heard of VX-Underground. They’re a legendary group sharing malware samples, research, and threat intel in a highly automated fashion. One thing that always impressed me about their ecosystem was the VX-U Telegram Bot that pushes timely updates—everything from ransomware leaks to threat advisories.

That got me thinking:
Why not build my own version for my research workflow?

In this article, I’ll walk you through how I built a fully automated Threat Intelligence bot that aggregates security news, ransomware leaks, and advisories from multiple sources, then sends real-time alerts to Telegram and Discord.

We’ll cover:

✔ Architecture
✔ RSS & JSON Parsing
✔ Deduplication & Logging
✔ Error Handling (fixing Discord 400 Bad Request)
✔ Deployment

By the end, you’ll have your own VX-Underground-inspired CTI bot running 24/7.

Why Build This Bot?

Threat Intelligence moves fast. New ransomware leaks, CVEs, and advisories drop every hour. Manually checking sources like CISA, NCSC, and security blogs is inefficient. Automation is key.

Our goal:
✅ Aggregate RSS feeds (blogs, advisories)
✅ Pull ransomware leak updates from JSON feeds
✅ Push to Telegram & Discord in real-time
✅ Avoid duplicates, broken URLs, and API rate limits

Architecture Overview

Here’s the architecture we built:

pgsqlCopyEdit        ┌───────────────────────────┐
        │       Data Sources        │
        │ RSS Feeds | JSON Feeds    │
        └───────────┬───────────────┘
                    │
            [Parser & Validator]
                    │
            [Deduplication Logic]
                    │
        ┌───────────┴───────────┐
        │ Discord Webhooks      │
        │ Telegram Bot API      │
        └───────────────────────┘

Sources:

RSS Feeds:
Ransomware Leak Data:
- https://raw.githubusercontent.com/joshhighet/ransomwatch/main/posts.json
- https://data.ransomware.live/posts.json

Tech Stack

Python 3.12
feedparser – RSS Parsing
aiohttp – Async HTTP requests
Telegram Bot API
Discord Webhooks
ConfigParser – For state tracking
logging – Structured debug logging

Phase 1: Fetching RSS Feeds

First, we set up our RSS parsing. Using feedparser, we can quickly grab articles from multiple sources.

pythonCopyEditimport feedparser
import logging
from urllib.parse import urlparse

logger = logging.getLogger(__name__)

async def get_news_from_rss(feed_urls):
    items = []
    for url in feed_urls:
        try:
            logger.debug(f"Parsing RSS feed: {url}")
            feed = feedparser.parse(url)
            for entry in feed.entries:
                # Validate URL
                if entry.link and is_valid_url(entry.link):
                    items.append({
                        "title": entry.title,
                        "link": entry.link,
                        "published": entry.get("published", "")
                    })
        except Exception as e:
            logger.error(f"Error parsing RSS feed {url}: {e}")
    return items

def is_valid_url(url):
    parsed = urlparse(url)
    return all([parsed.scheme in ["http", "https"], parsed.netloc])

✅ Fix implemented: urlparse() ensures no malformed URLs are sent to Discord.

Phase 2: Fetching Ransomware Data

Two JSON APIs give us ransomware group leak data.

pythonCopyEditimport aiohttp

async def get_ransomware_news():
    ransomware_feeds = [
        "https://raw.githubusercontent.com/joshhighet/ransomwatch/main/posts.json",
        "https://data.ransomware.live/posts.json"
    ]
    items = []
    async with aiohttp.ClientSession() as session:
        for feed_url in ransomware_feeds:
            try:
                async with session.get(feed_url) as resp:
                    if resp.status == 200:
                        data = await resp.json()
                        for post in data:
                            if "post_title" in post and "post_url" in post:
                                if is_valid_url(post["post_url"]):
                                    items.append({
                                        "title": post["post_title"],
                                        "link": post["post_url"],
                                        "group": post.get("group_name", "Unknown")
                                    })
            except Exception as e:
                logger.error(f"Error fetching ransomware feed {feed_url}: {e}")
    return items

Phase 3: Deduplication & Logging

We had a major issue: 15,000 old articles flooding Telegram every loop.
Fix: Maintain a processed set in memory + persist it in rss.log.

pythonCopyEditimport configparser
import os

PROCESSED_FILE = "rss.log"
processed_links = set()

def load_processed():
    if os.path.exists(PROCESSED_FILE):
        with open(PROCESSED_FILE, "r") as f:
            for line in f:
                processed_links.add(line.strip())

def save_processed(link):
    with open(PROCESSED_FILE, "a") as f:
        f.write(link + "\n")

Before sending each message:

pythonCopyEditif item["link"] not in processed_links:
    await send_to_discord(item)
    await send_to_telegram(item)
    processed_links.add(item["link"])
    save_processed(item["link"])

✅ This ensures only new articles get posted.

Phase 4: Sending to Telegram

We used python-telegram-bot for simplicity.

pythonCopyEditfrom telegram import Bot

TELEGRAM_TOKEN = "YOUR_TOKEN"
TELEGRAM_CHAT_ID = "YOUR_CHAT_ID"
bot = Bot(token=TELEGRAM_TOKEN)

async def send_to_telegram(item):
    text = f"📰 {item['title']}\n🔗 {item['link']}"
    await bot.send_message(chat_id=TELEGRAM_CHAT_ID, text=text)

Phase 5: Sending to Discord

We use Discord Webhooks with validation.

pythonCopyEditimport aiohttp

DISCORD_WEBHOOK = "YOUR_WEBHOOK_URL"

async def send_to_discord(item):
    embed = {
        "title": item["title"],
        "url": item["link"],
        "description": "New security update"
    }
    async with aiohttp.ClientSession() as session:
        async with session.post(DISCORD_WEBHOOK, json={"embeds": [embed]}) as resp:
            if resp.status != 204:
                logger.error(f"Discord error: {await resp.text()}")

✅ Fixed “Not a well-formed URL” issue by validating URLs before adding embed.url.

Deployment

Run it:

bashCopyEditpython rss.py

Or Dockerize:

dockerfileCopyEditFROM python:3.12
WORKDIR /app
COPY . /app
RUN pip install -r requirements.txt
CMD ["python", "rss.py"]

Add systemd service for persistence or use pm2.

Future Enhancements

✔ Integrate Dark Web scraping via Playwright + Tor
✔ IOC extraction → MISP/STIX format
✔ Discord slash commands (/threat latest)
✔ Web dashboard to view all intel