Инструменты и техники мониторинга Telegram-каналов в 2026 году

-каналы (публичные)

        │

        ▼

┌─────────────────────┐

│   Сборщик данных    │  ← Telethon / Pyrogram

│  (Collection Bot)   │

└──────────┬──────────┘

           │

           ▼

┌─────────────────────┐

│  База данных сырых  │  ← PostgreSQL / Elasticsearch

│      данных         │

└──────────┬──────────┘

           │

           ▼

┌─────────────────────┐

│  Процессор / Parser │  ← Python: извлечение IOC,

│                     │    ключевые слова, классификация

└──────────┬──────────┘

           │

           ▼

┌──────────────────────────────────────────────┐

│              Аналитический слой              │

│  ┌──────────┐  ┌──────────┐  ┌────────────┐ │

│  │  Kibana  │  │   MISP   │  │  Оповеще-  │ │

│  │ Grafana  │  │ OpenCTI  │  │ ния Slack/ │ │

│  │          │  │          │  │  Telegram  │ │

│  └──────────┘  └──────────┘  └────────────┘ │

└──────────────────────────────────────────────┘

<h2 id="shag-1-pereyti-na-my-telegram-org-apps">Шаг 1: Перейти на my.telegram.org/apps</h2>

<h2 id="shag-2-voyti-s-nomerom-telefona">Шаг 2: Войти с номером телефона</h2>

<h2 id="shag-3-sozdat-novoe-prilozhenie">Шаг 3: Создать новое приложение:</h2>

<h2 id="app-title-ti-monitoring-tool-lyuboe-nazvanie">App title: TI Monitoring Tool (любое название)</h2>

<h2 id="short-name-timonitor">Short name: timonitor</h2>

<h2 id="platform-other">Platform: Other</h2>

<h2 id="shag-4-sohranit-api-id-i-api-hash">Шаг 4: Сохранить api_id и api_hash</h2>




<h2 id="vazhno-ispolzovat-otdelnyy-akkaunt-dlya-monitoringa">Важно: использовать отдельный аккаунт для мониторинга,</h2>

<h2 id="ne-lichnyy-prichiny-opisany-v-razdele-opsec">не личный! Причины описаны в разделе OpSec.</h2>

<h2 id="ustanovka">Установка</h2>

pip install telethon



<h2 id="bazovyy-skript-podklyucheniya-i-chteniya-kanala">Базовый скрипт подключения и чтения канала</h2>

from telethon import TelegramClient, events

from telethon.tl.functions.channels import GetFullChannelRequest

import asyncio



API_ID = 12345678          # из my.telegram.org

API_HASH = "your_api_hash" # из my.telegram.org

SESSION_NAME = "ti_monitor"



async def main():

    async with TelegramClient(SESSION_NAME, API_ID, API_HASH) as client:



        # Получить информацию о канале

        channel = await client.get_entity("@channelname")

        full = await client(GetFullChannelRequest(channel))

        print(f"Канал: {channel.title}")

        print(f"Участников: {full.full_chat.participants_count}")



        # Получить последние 100 сообщений

        async for message in client.iter_messages(channel, limit=100):

            if message.text:

                print(f"[{message.date}] {message.text[:200]}")



asyncio.run(main())

<h2 id="ustanovka">Установка</h2>

pip install pyrogram tgcrypto



<h2 id="bazovyy-monitoring-novogo-soobscheniya-v-kanale">Базовый мониторинг нового сообщения в канале</h2>

from pyrogram import Client, filters



app = Client(

    "ti_monitor",

    api_id=12345678,

    api_hash="your_api_hash"

)



@app.on_message(filters.channel)

async def monitor_channel(client, message):

    """Обработчик каждого нового сообщения в любом подписанном канале"""

    print(f"Канал: {message.chat.title}")

    print(f"Время: {message.date}")

    print(f"Текст: {message.text}")

    # Здесь: сохранение в БД, проверка ключевых слов, извлечение IOC



app.run()

import requests



def send_alert(bot_token: str, chat_id: str, message: str):

    """

    Отправка оповещения аналитику через Telegram-бот.

    bot_token: токен от @BotFather

    chat_id: ID чата аналитика или группы ИБ-команды

    """

    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"

    payload = {

        "chat_id": chat_id,

        "text": message,

        "parse_mode": "HTML"

    }

    response = requests.post(url, json=payload)

    return response.json()



<h2 id="primer-vyzova-pri-obnaruzhenii-ugrozy">Пример вызова при обнаружении угрозы</h2>

alert_text = (

    "🚨 <b>ALERT: Обнаружен IOC</b>\n\n"

    f"Канал: @channel_name\n"

    f"IOC: 185.220.101.45\n"

    f"Контекст: продажа доступа к корпоративной сети\n"

    f"Время: 2026-02-15 14:32 UTC"

)

send_alert("BOT_TOKEN", "ANALYST_CHAT_ID", alert_text)

<h2 id="telegram-vstroennyy-poisk-v-kliente">Telegram встроенный поиск (в клиенте)</h2>

<h2 id="klyuchevye-slova-dlya-poiska-kanalov-ugroz">Ключевые слова для поиска каналов угроз:</h2>

<h2 id="access-for-sale-prodam-dostup">&quot;access for sale&quot; / &quot;продам доступ&quot;</h2>

<h2 id="database-leaked-sliv-bazy">&quot;database leaked&quot; / &quot;слив базы&quot;</h2>

<h2 id="combolist-combo">&quot;combolist&quot; / &quot;combo&quot;</h2>

<h2 id="stealer-logs-logs-shop">&quot;stealer logs&quot; / &quot;logs shop&quot;</h2>

<h2 id="ransomware-shifrovalschik">&quot;ransomware&quot; / &quot;шифровальщик&quot;</h2>

<h2 id="0day-0-day-exploit">&quot;0day&quot; / &quot;0-day exploit&quot;</h2>

<h2 id="initial-access-broker">&quot;initial access broker&quot;</h2>




<h2 id="intelx-intelligence-x-indeksiruet-telegram-kanaly">IntelX (Intelligence X) — индексирует Telegram-каналы</h2>

<h2 id="url-intelx-io">URL: intelx.io</h2>

<h2 id="poisk-po-klyuchevym-slovam-s-filtrom-source-telegram">Поиск по ключевым словам с фильтром source:telegram</h2>




<h2 id="telegago-poiskovik-po-telegram-kanalam">Telegago — поисковик по Telegram-каналам</h2>

<h2 id="url-telegago-com">URL: telegago.com</h2>




<h2 id="lyzem-poisk-po-kontentu-telegram-soobscheniy">Lyzem — поиск по контенту Telegram-сообщений</h2>

<h2 id="url-lyzem-com">URL: lyzem.com</h2>

from telethon import TelegramClient

from telethon.tl.functions.contacts import SearchRequest



async def search_channels(client, query: str, limit: int = 20):

    """

    Поиск каналов по ключевому слову через Telegram API.

    Документация: core.telegram.org/method/contacts.search

    """

    result = await client(SearchRequest(

        q=query,

        limit=limit

    ))



    channels = []

    for chat in result.chats:

        if hasattr(chat, 'broadcast') and chat.broadcast:

            channels.append({

                "id": chat.id,

                "title": chat.title,

                "username": getattr(chat, 'username', None),

                "participants": getattr(chat, 'participants_count', 0),

            })

            print(f"Найден: {chat.title} (@{getattr(chat, 'username', 'private')})"

                  f" — {getattr(chat, 'participants_count', '?')} подписчиков")

    return channels



<h2 id="spisok-poiskovyh-zaprosov-dlya-kanalov-ugroz">Список поисковых запросов для каналов угроз</h2>

threat_keywords = [

    "stealer logs", "combo list", "database leak",

    "initial access", "ransomware group", "exploit sale",

    "слив базы", "продам доступ", "взломанные аккаунты"

]

import re

from collections import Counter



async def extract_channel_mentions(client, channel, limit=500):

    """Извлечь все упоминания каналов из сообщений"""

    mentions = []

    async for msg in client.iter_messages(channel, limit=limit):

        if msg.text:

            # Паттерн для @username и t.me/username

            found = re.findall(

                r'@([a-zA-Z][a-zA-Z0-9_]{3,})|t\.me/([a-zA-Z][a-zA-Z0-9_]{3,})',

                msg.text

            )

            for match in found:

                username = match[0] or match[1]

                mentions.append(username.lower())



    # Топ упоминаемых каналов

    counter = Counter(mentions)

    return counter.most_common(20)

<h2 id="ti-collector-py-bazovyy-sborschik-soobscheniy">ti_collector.py — базовый сборщик сообщений</h2>

<h2 id="dokumentatsiya-telethon-docs-telethon-dev">Документация Telethon: docs.telethon.dev</h2>




import asyncio

import sqlite3

import json

from datetime import datetime

from telethon import TelegramClient, events

from telethon.tl.types import MessageMediaDocument, MessageMediaPhoto



API_ID = 12345678

API_HASH = "your_api_hash"



<h2 id="spisok-kanalov-dlya-monitoringa">Список каналов для мониторинга</h2>

CHANNELS = [

    "@channel1", "@channel2",  # добавьте свои каналы

]



def init_db():

    """Инициализация базы данных"""

    conn = sqlite3.connect("ti_monitor.db")

    conn.execute("""

        CREATE TABLE IF NOT EXISTS messages (

            id INTEGER PRIMARY KEY AUTOINCREMENT,

            message_id INTEGER,

            channel TEXT,

            channel_id INTEGER,

            text TEXT,

            date TEXT,

            has_media INTEGER,

            media_type TEXT,

            forward_from TEXT,

            collected_at TEXT

        )

    """)

    conn.commit()

    return conn



def save_message(conn, msg_data: dict):

    """Сохранить сообщение в БД"""

    conn.execute("""

        INSERT OR IGNORE INTO messages

        (message_id, channel, channel_id, text, date,

         has_media, media_type, forward_from, collected_at)

        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)

    """, (

        msg_data["message_id"], msg_data["channel"],

        msg_data["channel_id"], msg_data["text"],

        msg_data["date"], msg_data["has_media"],

        msg_data["media_type"], msg_data["forward_from"],

        datetime.utcnow().isoformat()

    ))

    conn.commit()



async def collect_history(client, channel_name: str, conn, limit=200):

    """Собрать историю сообщений канала"""

    try:

        entity = await client.get_entity(channel_name)

        count = 0

        async for msg in client.iter_messages(entity, limit=limit):

            if not msg.text and not msg.media:

                continue



            # Определить тип медиа

            media_type = None

            if isinstance(msg.media, MessageMediaDocument):

                media_type = "document"

            elif isinstance(msg.media, MessageMediaPhoto):

                media_type = "photo"



            # Форвард из другого канала

            forward_from = None

            if msg.forward and hasattr(msg.forward, 'chat'):

                forward_from = getattr(msg.forward.chat, 'username', None)



            msg_data = {

                "message_id": msg.id,

                "channel": channel_name,

                "channel_id": entity.id,

                "text": msg.text or "",

                "date": msg.date.isoformat(),

                "has_media": 1 if msg.media else 0,

                "media_type": media_type,

                "forward_from": forward_from,

            }

            save_message(conn, msg_data)

            count += 1



            await asyncio.sleep(0.05)  # Не превышать лимиты API



        print(f"✅ {channel_name}: собрано {count} сообщений")

    except Exception as e:

        print(f"❌ {channel_name}: ошибка — {e}")



async def realtime_monitor(client, conn):

    """Мониторинг новых сообщений в реальном времени"""

    @client.on(events.NewMessage(chats=CHANNELS))

    async def handler(event):

        msg = event.message

        channel = await event.get_chat()



        msg_data = {

            "message_id": msg.id,

            "channel": getattr(channel, 'username', str(channel.id)),

            "channel_id": channel.id,

            "text": msg.text or "",

            "date": msg.date.isoformat(),

            "has_media": 1 if msg.media else 0,

            "media_type": None,

            "forward_from": None,

        }

        save_message(conn, msg_data)

        print(f"📨 Новое: [{channel.title}] {msg.text[:100] if msg.text else '[медиа]'}")



    print("🔍 Мониторинг запущен. Ожидание новых сообщений...")

    await client.run_until_disconnected()



async def main():

    conn = init_db()

    async with TelegramClient("ti_session", API_ID, API_HASH) as client:

        # Сначала собрать историю

        for channel in CHANNELS:

            await collect_history(client, channel, conn)

            await asyncio.sleep(2)  # Пауза между каналами



        # Затем запустить мониторинг в реальном времени

        await realtime_monitor(client, conn)



asyncio.run(main())

<h2 id="import-sobrannyh-soobscheniy-v-elasticsearch">Импорт собранных сообщений в Elasticsearch</h2>

<h2 id="trebuet-pip-install-elasticsearch">Требует: pip install elasticsearch</h2>

from elasticsearch import Elasticsearch

import sqlite3

import json



es = Elasticsearch("http://localhost:9200")



def export_to_elasticsearch():

    conn = sqlite3.connect("ti_monitor.db")

    conn.row_factory = sqlite3.Row

    cursor = conn.execute("SELECT * FROM messages ORDER BY date DESC")



    for row in cursor:

        doc = dict(row)

        doc["@timestamp"] = doc.pop("date")  # Kibana ожидает @timestamp



        es.index(

            index="ti-telegram-messages",

            id=f"{doc['channel']}_{doc['message_id']}",

            document=doc

        )



    print("✅ Данные экспортированы в Elasticsearch")

    conn.close()



export_to_elasticsearch()

<h2 id="docker-compose-yml-dlya-steka-elk">docker-compose.yml для стека ELK</h2>

version: '3.8'

services:

  elasticsearch:

    image: elasticsearch:8.12.0

    environment:

      - discovery.type=single-node

      - xpack.security.enabled=false

      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"

    ports:

      - "9200:9200"

    volumes:

      - esdata:/usr/share/elasticsearch/data



  kibana:

    image: kibana:8.12.0

    ports:

      - "5601:5601"

    environment:

      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200

    depends_on:

      - elasticsearch



volumes:

  esdata:

<h2 id="zapusk-steka">Запуск стека</h2>

docker-compose up -d



<h2 id="proverka-dostupnosti">Проверка доступности</h2>

curl http://localhost:9200/_cluster/health

<h2 id="kibana-http-localhost-5601">Kibana: http://localhost:5601</h2>

<h2 id="threat-keywords-py-slovar-klyuchevyh-slov-dlya-monitoringa-ugroz">threat_keywords.py — словарь ключевых слов для мониторинга угроз</h2>




THREAT_KEYWORDS = {



    "initial_access": [

        # Продажа корпоративных доступов

        "продам доступ", "access for sale", "corporate access",

        "rdp access", "vpn access", "citrix", "initial access broker",

        "IAB", "network access", "доступ к сети", "шелл", "webshell",

        "shell access", "c-panel access", "cpanel",

    ],



    "data_leaks": [

        # Утечки данных и базы

        "слив базы", "база данных", "database leak", "data breach",

        "combolist", "combo list", "credential stuffing",

        "leaked database", "dump", "дамп", "утечка",

        "персональные данные", "пробив", "пасспорт слив",

    ],



    "malware": [

        # Вредоносное ПО

        "stealer", "стилер", "infostealer", "rat", "trojan",

        "ransomware", "шифровальщик", "locker", "crypter",

        "loader", "dropper", "botnet", "ботнет", "malware for sale",

        "stealer logs", "logs shop", "фуд", "fud",

    ],



    "exploits": [

        # Эксплойты и уязвимости

        "0day", "0-day", "exploit", "эксплойт", "poc exploit",

        "rce exploit", "lpe exploit", "cve-2025", "cve-2026",

        "privilege escalation", "bypass", "уязвимость",

        "продам эксплойт", "exploit sale",

    ],



    "phishing": [

        # Фишинг и мошенничество

        "фишинг", "phishing kit", "phish", "landing page",

        "фишинговая страница", "carding", "кардинг",

        "drops needed", "дропы", "money mule",

    ],



    "ddos": [

        # DDoS услуги

        "ddos", "ддос", "stresser", "booter", "l7 ddos",

        "l4 ddos", "hire ddos", "ddos service", "загружу сайт",

    ],

}



<h2 id="vysokoprioritetnye-terminy-nemedlennoe-opoveschenie">Высокоприоритетные термины (немедленное оповещение)</h2>

HIGH_PRIORITY = [

    "critical infrastructure", "атомная", "электростанция",

    "government access", "доступ к госслужбам",

    "финансовая организация", "банк доступ",

    "нулевой день", "zero day rce",

]

<h2 id="ioc-extractor-py-izvlechenie-indikatorov-komprometatsii">ioc_extractor.py — извлечение индикаторов компрометации</h2>

import re

from typing import Dict, List



class IOCExtractor:

    """

    Извлечение IOC (индикаторов компрометации) из текста.

    Поддерживает: IP, домены, URL, хеши файлов, CVE, email

    """



    PATTERNS = {

        "ipv4": r'\b(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}'

                r'(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b',



        "domain": r'\b(?:<a href="?:[a-zA-Z0-9\-]{0,61}[a-zA-Z0-9]">a-zA-Z0-9</a>?\.)'

                  r'+(?:com|net|org|ru|io|co|info|biz|xyz|top|pw|tk|ml)\b',



        "url": r'https?://[^\s<>"{}|\\^`\[\]]{10,}',



        "md5": r'\b[a-fA-F0-9]{32}\b',

        "sha1": r'\b[a-fA-F0-9]{40}\b',

        "sha256": r'\b[a-fA-F0-9]{64}\b',



        "cve": r'CVE-\d{4}-\d{4,7}',



        "email": r'\b[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}\b',



        "bitcoin_address": r'\b[13][a-km-zA-HJ-NP-Z1-9]{25,34}\b',

        "monero_address": r'\b4[0-9AB][1-9A-HJ-NP-Za-km-z]{93}\b',

    }



    def extract(self, text: str) -> Dict[str, List[str]]:

        """Извлечь все IOC из текста"""

        results = {}

        for ioc_type, pattern in self.PATTERNS.items():

            matches = re.findall(pattern, text)

            # Дедупликация и фильтрация ложных срабатываний

            unique = list(set(matches))

            if unique:

                results[ioc_type] = unique

        return results



    def extract_with_context(self, text: str, context_chars: int = 100) -> List[dict]:

        """Извлечь IOC с контекстом — фрагментом окружающего текста"""

        iocs_with_context = []

        for ioc_type, pattern in self.PATTERNS.items():

            for match in re.finditer(pattern, text):

                start = max(0, match.start() - context_chars)

                end = min(len(text), match.end() + context_chars)

                iocs_with_context.append({

                    "type": ioc_type,

                    "value": match.group(),

                    "context": text[start:end]

                })

        return iocs_with_context



<h2 id="ispolzovanie">Использование</h2>

extractor = IOCExtractor()



sample_text = """

Продам доступ RDP к корпоративной сети.

IP сервера: 185.220.101.45

Стилер детектируется на: abc123def456abc123def456abc123de (MD5)

Уязвимость: CVE-2025-44228 — активная эксплуатация

Связь: hacker@protonmail.com

"""



iocs = extractor.extract(sample_text)

print(iocs)

<h2 id="ipv4-185-220-101-45-md5-abc123-cve-cve-2025-44228">{&#039;ipv4&#039;: [&#039;185.220.101.45&#039;], &#039;md5&#039;: [&#039;abc123...&#039;], &#039;cve&#039;: [&#039;CVE-2025-44228&#039;], ...}</h2>

from threat_keywords import THREAT_KEYWORDS, HIGH_PRIORITY



def classify_message(text: str) -> dict:

    """

    Классифицировать сообщение по типу угрозы и приоритету.

    Возвращает: категория, приоритет (1-3), совпавшие ключевые слова

    """

    text_lower = text.lower()

    matches = {}



    for category, keywords in THREAT_KEYWORDS.items():

        found = [kw for kw in keywords if kw.lower() in text_lower]

        if found:

            matches[category] = found



    # Определить приоритет

    priority = 1  # Низкий по умолчанию

    if any(hp.lower() in text_lower for hp in HIGH_PRIORITY):

        priority = 3  # Критический

    elif matches:

        priority = 2  # Средний — есть совпадения по категориям



    return {

        "categories": list(matches.keys()),

        "matched_keywords": matches,

        "priority": priority,

        "is_threat": len(matches) > 0

    }

<h2 id="primer-ispolzovaniya-tgstat-api-trebuet-klyuch">Пример использования TGStat API (требует ключ)</h2>

import requests



TGSTAT_TOKEN = "your_tgstat_token"



def search_tgstat(query: str, limit: int = 20):

    """Поиск сообщений по ключевому слову через TGStat API"""

    url = "https://api.tgstat.ru/posts/search"

    params = {

        "token": TGSTAT_TOKEN,

        "q": query,

        "limit": limit,

        "extended": 1

    }

    response = requests.get(url, params=params)

    return response.json()



<h2 id="poisk-upominaniy-ioc">Поиск упоминаний IOC</h2>

results = search_tgstat("CVE-2025-44228 exploit sale")

<h2 id="brand-monitor-py-monitoring-upominaniy-brenda">brand_monitor.py — мониторинг упоминаний бренда</h2>




<h2 id="nastroyte-pod-svoyu-organizatsiyu">Настройте под свою организацию</h2>

BRAND_TERMS = [

    "yourcompany.ru",       # Домен

    "yourcompany",          # Название

    "@yourcompany_ru",      # Telegram-аккаунты

    "ваша компания",        # Бренд на русском

    "yourcompany VPN",      # Продукты

    "yourcompany admin",    # Комбинации с ролями

    "ceo@yourcompany.ru",   # Корпоративные адреса топ-менеджмента

]



def check_brand_mention(text: str) -> list:

    """Проверить текст на упоминание контролируемых активов"""

    text_lower = text.lower()

    found = []

    for term in BRAND_TERMS:

        if term.lower() in text_lower:

            found.append(term)

    return found



<h2 id="v-obrabotchike-novyh-soobscheniy">В обработчике новых сообщений:</h2>

async def handler(event):

    msg = event.message

    if msg.text:

        brand_hits = check_brand_mention(msg.text)

        if brand_hits:

            # НЕМЕДЛЕННОЕ оповещение!

            alert = (

                f"🚨 БРЕНД МОНИТОРИНГ\n\n"

                f"Совпадения: {', '.join(brand_hits)}\n"

                f"Канал: {event.chat.title}\n"

                f"Текст: {msg.text[:500]}"

            )

            await send_alert(ANALYST_BOT_TOKEN, ANALYST_CHAT_ID, alert)

<h2 id="tipichnye-patterny-obyavleniy-iab">Типичные паттерны объявлений IAB</h2>

IAB_PATTERNS = {

    "rdp_access": [

        r"rdp.*(?:admin|administrator).*(?:доступ|access|sale)",

        r"(?:продам|sell|for sale).*rdp.*(?:server|сервер)",

    ],

    "vpn_access": [

        r"(?:cisco|fortinet|pulse|palo alto).*(?:vpn|access)",

        r"(?:продам|sell).*vpn.*(?:corporate|корпоратив)",

    ],

    "revenue_indicator": [

        r"\$[\d,]+(?:k|m|млн|тыс)?.*(?:revenue|выручка|оборот)",

        r"(?:company size|сотрудников|employees).*\d+",

    ],

}



import re



def detect_iab_pattern(text: str) -> list:

    """Обнаружить паттерны объявлений IAB"""

    text_lower = text.lower()

    detected = []

    for pattern_type, patterns in IAB_PATTERNS.items():

        for pattern in patterns:

            if re.search(pattern, text_lower):

                detected.append(pattern_type)

                break

    return detected

<h2 id="kanaly-dlya-monitoringa-utechek-dannyh-primery-kategoriy">Каналы для мониторинга утечек данных (примеры категорий)</h2>

<h2 id="aktualnyy-spisok-podderzhivayte-na-osnove-sobstvennogo-issledovaniya">Актуальный список поддерживайте на основе собственного исследования</h2>




LEAK_MONITORING_CATEGORIES = [

    "combolist",        # Каналы с combo-листами

    "database_leaks",   # Утечки корпоративных баз

    "stealer_logs",     # Логи стилеров с корпоративными аккаунтами

    "credential_dumps", # Дампы учётных данных

]



async def process_leak_message(msg_text: str, channel_name: str):

    """Обработка сообщений из каналов утечек"""

    extractor = IOCExtractor()

    brand_hits = check_brand_mention(msg_text)

    iocs = extractor.extract(msg_text)



    if brand_hits:

        # Немедленное уведомление при упоминании нашей организации

        urgency = "🆘 КРИТИЧНО"

        priority = 3

    elif iocs.get("email") or iocs.get("domain"):

        urgency = "⚠️ ВАЖНО"

        priority = 2

    else:

        urgency = "ℹ️ ИНФО"

        priority = 1



    return {

        "urgency": urgency,

        "priority": priority,

        "brand_mentions": brand_hits,

        "iocs": iocs,

        "channel": channel_name

    }

<h2 id="channel-graph-py-postroenie-grafa-svyazey-na-osnove-repostov-i-upominaniy">channel_graph.py — построение графа связей на основе репостов и упоминаний</h2>

<h2 id="trebuet-pip-install-networkx-matplotlib">Требует: pip install networkx matplotlib</h2>




import networkx as nx

import matplotlib.pyplot as plt

from collections import defaultdict



class ChannelGraph:

    """Граф связей между Telegram-каналами на основе репостов"""



    def __init__(self):

        self.graph = nx.DiGraph()

        self.forward_counts = defaultdict(lambda: defaultdict(int))



    def add_forward(self, source_channel: str, forward_from: str):

        """Добавить рёбро: channel репостит из forward_from"""

        self.forward_counts[source_channel][forward_from] += 1

        weight = self.forward_counts[source_channel][forward_from]

        self.graph.add_edge(forward_from, source_channel, weight=weight)



    def add_mention(self, source_channel: str, mentioned_channel: str):

        """Добавить рёбро: channel упоминает mentioned_channel"""

        if not self.graph.has_edge(source_channel, mentioned_channel):

            self.graph.add_edge(source_channel, mentioned_channel,

                                weight=1, edge_type="mention")



    def find_clusters(self):

        """Найти кластеры связанных каналов"""

        undirected = self.graph.to_undirected()

        clusters = list(nx.connected_components(undirected))

        return sorted(clusters, key=len, reverse=True)



    def find_central_channels(self, top_n: int = 10):

        """Найти наиболее центральные каналы (хабы экосистемы угроз)"""

        centrality = nx.degree_centrality(self.graph)

        return sorted(centrality.items(), key=lambda x: -x[1])[:top_n]



    def visualize(self, output_file: str = "channel_graph.png"):

        """Визуализировать граф связей"""

        plt.figure(figsize=(20, 15))

        pos = nx.spring_layout(self.graph, k=2, iterations=50)

        nx.draw_networkx(

            self.graph, pos,

            node_size=500,

            font_size=8,

            arrows=True,

            edge_color="gray",

            node_color="lightblue"

        )

        plt.title("Граф связей Telegram-каналов угроз")

        plt.savefig(output_file, dpi=150, bbox_inches='tight')

        print(f"✅ Граф сохранён: {output_file}")

import hashlib



def create_writing_fingerprint(messages: list) -> dict:

    """

    Создать fingerprint стиля написания по набору сообщений.

    Основан на статистических характеристиках текста.

    """

    if not messages:

        return {}



    all_text = " ".join(messages)

    words = all_text.lower().split()



    # Характеристики стиля

    fingerprint = {

        "avg_message_length": sum(len(m) for m in messages) / len(messages),

        "emoji_frequency": sum(1 for c in all_text if ord(c) > 127000) / len(messages),

        "caps_ratio": sum(1 for c in all_text if c.isupper()) / max(len(all_text), 1),

        "unique_word_ratio": len(set(words)) / max(len(words), 1),

    }

    return fingerprint

def extract_contact_infrastructure(text: str) -> dict:

    """Извлечь инфраструктурные маркеры для отслеживания актора"""

    extractor = IOCExtractor()

    iocs = extractor.extract(text)



    # Дополнительный поиск контактных данных

    tg_contacts = re.findall(r'@([a-zA-Z][a-zA-Z0-9_]{4,})', text)

    tox_ids = re.findall(r'[A-F0-9]{76}', text)  # TOX ID

    jabber = re.findall(r'[a-zA-Z0-9._%+\-]+@(?:jabber|xmpp)\.[a-zA-Z]+', text)



    return {

        "crypto": {

            "bitcoin": iocs.get("bitcoin_address", []),

            "monero": iocs.get("monero_address", []),

        },

        "contacts": {

            "telegram": tg_contacts,

            "tox": tox_ids,

            "jabber": jabber,

        }

    }

import sqlite3

from datetime import datetime, timedelta

import collections



def analyze_activity_pattern(channel_name: str, days: int = 30) -> dict:

    """

    Анализ паттерна активности канала:

    когда публикует, как часто, в какие часовые пояса.

    Позволяет определить географию и режим работы актора.

    """

    conn = sqlite3.connect("ti_monitor.db")

    since = (datetime.utcnow() - timedelta(days=days)).isoformat()



    cursor = conn.execute("""

        SELECT date FROM messages

        WHERE channel = ? AND date >= ?

        ORDER BY date

    """, (channel_name, since))



    timestamps = [datetime.fromisoformat(row[0]) for row in cursor]

    conn.close()



    if not timestamps:

        return {"error": "No data"}



    # Распределение по часам UTC

    hour_distribution = collections.Counter(ts.hour for ts in timestamps)

    # Распределение по дням недели (0=понедельник)

    weekday_distribution = collections.Counter(ts.weekday() for ts in timestamps)



    # Определить пиковый час активности

    peak_hour = max(hour_distribution, key=hour_distribution.get)



    return {

        "total_messages": len(timestamps),

        "messages_per_day": len(timestamps) / days,

        "peak_hour_utc": peak_hour,

        "hourly_distribution": dict(sorted(hour_distribution.items())),

        "weekday_distribution": dict(sorted(weekday_distribution.items())),

    }

<h2 id="misp-integration-py">misp_integration.py</h2>

<h2 id="trebuet-pip-install-pymisp">Требует: pip install pymisp</h2>




from pymisp import PyMISP, MISPEvent, MISPAttribute



MISP_URL = "https://your-misp-instance.com"

MISP_KEY = "your-misp-api-key"



def create_misp_event_from_telegram(

    channel: str,

    message_text: str,

    iocs: dict,

    threat_category: str

) -> str:

    """

    Создать событие в MISP на основе находки из Telegram-мониторинга.

    Документация PyMISP: pymisp.readthedocs.io

    """

    misp = PyMISP(MISP_URL, MISP_KEY, False)



    event = MISPEvent()

    event.info = f"TI Telegram Monitor: {threat_category} in {channel}"

    event.threat_level_id = 2  # Medium

    event.distribution = 0     # Only your organisation

    event.analysis = 1         # Ongoing



    # Добавить атрибуты IOC

    if "ipv4" in iocs:

        for ip in iocs["ipv4"]:

            attr = MISPAttribute()

            attr.type = "ip-dst"

            attr.value = ip

            attr.comment = f"From Telegram channel {channel}"

            event.add_attribute(attr)



    if "domain" in iocs:

        for domain in iocs["domain"]:

            event.add_attribute("domain", domain,

                                comment=f"Telegram: {channel}")



    if "sha256" in iocs:

        for hash_val in iocs["sha256"]:

            event.add_attribute("sha256", hash_val,

                                comment=f"Telegram: {channel}")



    if "cve" in iocs:

        for cve in iocs["cve"]:

            event.add_attribute("vulnerability", cve,

                                comment=f"Telegram: {channel}")



    result = misp.add_event(event)

    return result["Event"]["uuid"]

<h2 id="opencti-integration-py">opencti_integration.py</h2>

<h2 id="trebuet-pip-install-pycti">Требует: pip install pycti</h2>




from pycti import OpenCTIApiClient



OPENCTI_URL = "http://localhost:8080"

OPENCTI_TOKEN = "your-opencti-token"



def push_indicator_to_opencti(ioc_value: str, ioc_type: str,

                               context: str, channel: str):

    """

    Добавить индикатор в OpenCTI.

    Документация: docs.opencti.io

    """

    client = OpenCTIApiClient(OPENCTI_URL, OPENCTI_TOKEN)



    # Создать индикатор

    indicator = client.indicator.create(

        name=f"Telegram TI: {ioc_value}",

        description=f"Обнаружен в канале {channel}. Контекст: {context[:200]}",

        pattern=f"[{ioc_type}:value = '{ioc_value}']",

        pattern_type="stix",

        x_opencti_score=75,

        confidence=50,

    )

    return indicator

<h2 id="alert-dispatcher-py-marshrutizatsiya-opovescheniy-po-kanalam">alert_dispatcher.py — маршрутизация оповещений по каналам</h2>




import requests

import json



def send_slack_alert(webhook_url: str, alert: dict):

    """Отправка структурированного оповещения в Slack"""

    color = {"3": "danger", "2": "warning", "1": "good"}[str(alert["priority"])]



    payload = {

        "attachments": [{

            "color": color,

            "title": f"🛡️ TI Alert: {alert.get('threat_category', 'Угроза')}",

            "fields": [

                {"title": "Канал", "value": alert["channel"], "short": True},

                {"title": "Приоритет", "value": str(alert["priority"]), "short": True},

                {"title": "IOC", "value": json.dumps(alert.get("iocs", {}), ensure_ascii=False)},

                {"title": "Контекст", "value": alert.get("context", "")[:500]},

            ],

            "footer": "Telegram TI Monitor",

            "ts": alert.get("timestamp")

        }]

    }

    requests.post(webhook_url, json=payload)



<h2 id="otpravka-v-siem-cherez-syslog-cef-naprimer-v-splunk-ili-qradar">Отправка в SIEM через Syslog/CEF (например, в Splunk или QRadar)</h2>

import socket



def send_to_siem_syslog(siem_host: str, siem_port: int, alert: dict):

    """Отправить событие в SIEM через UDP Syslog"""

    cef_message = (

        f"CEF:0|TI-Monitor|TelegramMonitor|1.0|TI_ALERT|"

        f"Telegram Threat Intelligence Alert|{alert['priority']}|"

        f"channel={alert['channel']} "

        f"ioc_count={len(alert.get('iocs', {}))} "

        f"category={alert.get('threat_category', 'unknown')}"

    )

    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    sock.sendto(cef_message.encode(), (siem_host, siem_port))

    sock.close()

<h2 id="zapusk-monitoringa-cherez-vpn-tor-dvoynaya-anonimizatsiya">Запуск мониторинга через VPN + Tor (двойная анонимизация)</h2>

<h2 id="ispolzovat-dlya-osobo-chuvstvitelnyh-kanalov">Использовать для особо чувствительных каналов</h2>




<h2 id="variant-1-torsocks-dlya-suschestvuyuschih-skriptov">Вариант 1: Torsocks для существующих скриптов</h2>

torsocks python3 ti_collector.py



<h2 id="variant-2-zapusk-cherez-docker-s-tor-proksi">Вариант 2: Запуск через Docker с Tor-прокси</h2>

<h2 id="docker-compose-yml-s-tor-servisom">docker-compose.yml с Tor-сервисом</h2>

<h2 id="vse-zaprosy-marshrutizirovat-cherez-socks5-tor-9050">Все запросы маршрутизировать через socks5://tor:9050</h2>




<h2 id="dedicated-vm-dlya-monitoringa-izolyatsiya-ot-osnovnoy-infrastruktury">Dedicated VM для мониторинга (изоляция от основной инфраструктуры)</h2>

<h2 id="rekomenduemaya-konfiguratsiya">Рекомендуемая конфигурация:</h2>

<h2 id="otdelnaya-vm-tolko-dlya-monitoringa">- Отдельная VM только для мониторинга</h2>

<h2 id="ne-ispolzovat-dlya-drugih-zadach">- Не использовать для других задач</h2>

<h2 id="snapshot-chistogo-sostoyaniya-dlya-vosstanovleniya">- Снапшот чистого состояния для восстановления</h2>

<h2 id="konfiguratsiya-telethon-s-kastomnym-device-model">Конфигурация Telethon с кастомным device_model</h2>

from telethon import TelegramClient



<h2 id="imitirovat-standartnyy-mobilnyy-klient-a-ne-bot">Имитировать стандартный мобильный клиент, а не &quot;бот&quot;</h2>

client = TelegramClient(

    "ti_session",

    API_ID,

    API_HASH,

    device_model="Samsung Galaxy S23",

    system_version="Android 14",

    app_version="10.3.2",

    lang_code="ru",

    system_lang_code="ru-RU"

)