July 7
Автоматизация технического OSINT: сбор данных в скриптах
Автоматизация технического OSINT: сбор данных в скриптах (Python, Bash)
Ручное выполнение всех этапов технического OSINT для одного домена может занять несколько часов. При необходимости мониторинга десятков или сотен целей ручной сбор становится невозможным. Автоматизация превращает разрозненные запросы к API, парсинг веб-страниц и анализ данных в единый воспроизводимый конвейер, который можно запускать по расписанию. Правильно построенная система автоматизации позволяет получать актуальную информацию о целях без участия человека, экономить время и избегать ошибок. Данная статья посвящена методам и практическим примерам автоматизации технического OSINT с использованием Python и Bash.
Зачем автоматизировать технический OSINT?
Рекомендуемая структура каталога для автоматизации:
osint-automation/ ├── config/ │ ├── config.yaml # API ключи, настройки │ └── targets.txt # Список доменов (по одному на строку) ├── modules/ │ ├── dns.py # Сбор DNS-записей │ ├── whois.py # WHOIS-информация │ ├── subdomains.py # Поиск поддоменов (crt.sh, SecurityTrails) │ ├── certificates.py # SSL/TLS сертификаты │ ├── shodan.py # Shodan-данные │ ├── tech_stack.py # Wappalyzer/BuiltWith технологии │ ├── github.py # Поиск секретов в GitHub │ └── changes.py # Сравнение с предыдущим состоянием ├── utils/ │ ├── logger.py # Логирование │ ├── notifier.py # Отправка уведомлений (Telegram, Email) │ └── db.py # Работа с SQLite ├── data/ # Хранилище результатов (JSON, SQLite) ├── main.py # Главный скрипт ├── requirements.txt # Зависимости Python └── run.sh # Bash-обёртка для cron
Этап 2: Настройка конфигурации (config.yaml)
# config.yaml
api_keys:
shodan: "SHODAN_API_KEY"
securitytrails: "SECURITYTRAILS_API_KEY"
github: "GITHUB_TOKEN"
virustotal: "VIRUSTOTAL_API_KEY" # опционально
censys:
api_id: "CENSYS_API_ID"
api_secret: "CENSYS_API_SECRET"
notifications:
telegram:
token: "TELEGRAM_BOT_TOKEN"
chat_id: "TELEGRAM_CHAT_ID"
email:
smtp_server: "smtp.gmail.com"
smtp_port: 587
username: "user@gmail.com"
password: "app_password"
to: "recipient@example.com"
database:
path: "data/osint.db"
targets:
file: "config/targets.txt"
logging:
level: "INFO"
file: "data/osint.log"
scan:
# интервал повторных запросов для Shodan (если нужно ждать)
shodan_delay: 1 # секунды между запросами
timeout: 30 # таймаут для HTTP-запросовЭтап 3: Сбор DNS-записей (modules/dns.py)
import dns.resolver
import json
from utils.logger import get_logger
logger = get_logger(__name__)
def get_dns_records(domain):
"""
Сбор основных DNS-записей: A, AAAA, MX, TXT, NS, CNAME, SOA
"""
records = {}
record_types = ['A', 'AAAA', 'MX', 'TXT', 'NS', 'CNAME', 'SOA']
for rtype in record_types:
try:
answers = dns.resolver.resolve(domain, rtype)
records[rtype] = [str(r) for r in answers]
except dns.resolver.NoAnswer:
records[rtype] = []
except dns.resolver.NXDOMAIN:
records[rtype] = []
except Exception as e:
logger.warning(f"DNS {rtype} для {domain}: {e}")
records[rtype] = []
return records
def get_reverse_dns(ip):
"""
Обратный DNS (PTR) для IP-адреса
"""
try:
answers = dns.resolver.resolve_address(ip)
return [str(r) for r in answers]
except:
return []
# Пример использования
if __name__ == "__main__":
records = get_dns_records("example.com")
print(json.dumps(records, indent=2))Этап 4: WHOIS-информация (modules/whois.py)
import whois
import json
from datetime import datetime
def get_whois_info(domain):
"""
Получение WHOIS-информации о домене
"""
try:
w = whois.whois(domain)
return {
'domain_name': w.domain_name,
'registrar': w.registrar,
'creation_date': str(w.creation_date),
'expiration_date': str(w.expiration_date),
'updated_date': str(w.updated_date),
'name_servers': w.name_servers,
'emails': w.emails,
'org': w.org,
'country': w.country,
}
except Exception as e:
return {'error': str(e)}Этап 5: Поиск поддоменов через crt.sh (modules/subdomains.py)
import requests
import json
from utils.logger import get_logger
logger = get_logger(__name__)
def get_subdomains_crtsh(domain):
"""
Получение поддоменов из Certificate Transparency логов через crt.sh
"""
url = f"https://crt.sh/?q=%25.{domain}&output=json"
subdomains = set()
try:
resp = requests.get(url, timeout=30)
if resp.status_code == 200:
data = resp.json()
for entry in data:
name = entry.get('name_value', '')
if name.endswith(f'.{domain}') or name == domain:
# убираем wildcard
if name.startswith('*.'):
name = name[2:]
subdomains.add(name.lower())
else:
logger.warning(f"crt.sh ошибка: {resp.status_code}")
except Exception as e:
logger.error(f"crt.sh исключение: {e}")
return sorted(subdomains)
def get_subdomains_securitytrails(domain, api_key):
"""
Получение поддоменов через SecurityTrails API (требуется ключ)
"""
if not api_key:
return []
url = f"https://api.securitytrails.com/v1/domain/{domain}/subdomains"
headers = {"APIKEY": api_key}
try:
resp = requests.get(url, headers=headers, timeout=30)
if resp.status_code == 200:
data = resp.json()
subdomains = data.get('subdomains', [])
return [f"{s}.{domain}" for s in subdomains]
except Exception as e:
logger.error(f"SecurityTrails ошибка: {e}")
return []
def get_all_subdomains(domain, securitytrails_key=None):
"""
Агрегация поддоменов из всех источников
"""
all_subs = set()
all_subs.update(get_subdomains_crtsh(domain))
if securitytrails_key:
all_subs.update(get_subdomains_securitytrails(domain, securitytrails_key))
return sorted(all_subs)Этап 6: Анализ SSL/TLS сертификатов (modules/certificates.py)
import requests
import json
from datetime import datetime
def get_certificates_crtsh(domain):
"""
Получение информации о сертификатах для домена
"""
url = f"https://crt.sh/?q=%25.{domain}&output=json"
certs = []
try:
resp = requests.get(url, timeout=30)
if resp.status_code == 200:
data = resp.json()
for entry in data:
if len(entry) < 6:
continue
certs.append({
'name': entry[1],
'issuer': entry[2],
'valid_from': entry[3],
'valid_to': entry[4],
'ca': entry[5]
})
except Exception as e:
pass
return certs
def get_certificate_expiry(certs):
"""
Определение ближайшего истекающего сертификата
"""
now = datetime.now()
expiring = []
for cert in certs:
try:
valid_to = datetime.strptime(cert['valid_to'], '%Y-%m-%dT%H:%M:%S')
days_left = (valid_to - now).days
if days_left < 30:
expiring.append({
'name': cert['name'],
'expires': cert['valid_to'],
'days_left': days_left
})
except:
pass
return expiringЭтап 7: Сбор данных из Shodan (modules/shodan.py)
import shodan
from utils.logger import get_logger
logger = get_logger(__name__)
def get_shodan_host(ip, api_key):
"""
Получение информации о хосте из Shodan
"""
if not api_key:
return {}
try:
api = shodan.Shodan(api_key)
host = api.host(ip)
return {
'ip': host.get('ip_str'),
'organization': host.get('org'),
'isp': host.get('isp'),
'country': host.get('country_name'),
'ports': host.get('ports', []),
'vulns': host.get('vulns', []),
'data': host.get('data', []) # баннеры
}
except shodan.APIError as e:
logger.error(f"Shodan ошибка для {ip}: {e}")
return {}
except Exception as e:
logger.error(f"Shodan исключение для {ip}: {e}")
return {}
def get_shodan_host_by_domain(domain, api_key):
"""
Поиск хостов, связанных с доменом, через Shodan
"""
if not api_key:
return []
try:
api = shodan.Shodan(api_key)
results = api.search(f"hostname:{domain}")
hosts = []
for match in results.get('matches', []):
hosts.append({
'ip': match.get('ip_str'),
'port': match.get('port'),
'data': match.get('data', '')[:200]
})
return hosts
except shodan.APIError as e:
logger.error(f"Shodan search ошибка: {e}")
return []Этап 8: Поиск в GitHub (modules/github.py)
import requests
from utils.logger import get_logger
logger = get_logger(__name__)
def search_github_secrets(domain, token):
"""
Поиск упоминаний домена и подозрительных паттернов в GitHub
"""
if not token:
return []
# Комбинированный запрос: упоминания домена + файлы .env, пароли, ключи
queries = [
f'"{domain}" extension:env',
f'"{domain}" password',
f'"{domain}" api_key',
f'"{domain}" secret',
f'"{domain}" token',
]
headers = {
"Authorization": f"token {token}",
"Accept": "application/vnd.github.v3+json"
}
all_results = []
for query in queries:
url = "https://api.github.com/search/code"
params = {"q": query, "per_page": 20}
try:
resp = requests.get(url, headers=headers, params=params, timeout=20)
if resp.status_code == 200:
data = resp.json()
for item in data.get('items', []):
all_results.append({
'repo': item['repository']['full_name'],
'path': item['path'],
'url': item['html_url'],
'query': query
})
else:
logger.warning(f"GitHub ошибка ({query}): {resp.status_code}")
except Exception as e:
logger.error(f"GitHub исключение: {e}")
return all_resultsЭтап 9: Технологический стек (modules/tech_stack.py)
import subprocess
import json
from utils.logger import get_logger
logger = get_logger(__name__)
def get_tech_wappalyzer(url):
"""
Определение технологий через Wappalyzer CLI
Требуется установить: npm install -g wappalyzer
"""
try:
result = subprocess.run(
['wappalyzer', url],
capture_output=True,
text=True,
timeout=60
)
if result.returncode == 0:
return json.loads(result.stdout)
else:
logger.warning(f"Wappalyzer ошибка: {result.stderr}")
return {}
except Exception as e:
logger.error(f"Wappalyzer исключение: {e}")
return {}
def get_tech_builtwith(url, api_key=None):
"""
Использование BuiltWith API (требуется ключ)
"""
if not api_key:
return {}
endpoint = f"https://api.builtwith.com/free1/api.json?KEY={api_key}&LOOKUP={url}"
try:
import requests
resp = requests.get(endpoint, timeout=30)
if resp.status_code == 200:
return resp.json()
except Exception as e:
logger.error(f"BuiltWith ошибка: {e}")
return {}Этап 10: Общая база данных (utils/db.py)
Используем SQLite для хранения истории изменений.
import sqlite3
import json
from datetime import datetime
class OSINTDatabase:
def __init__(self, db_path):
self.conn = sqlite3.connect(db_path)
self.create_tables()
def create_tables(self):
cursor = self.conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS domains (
id INTEGER PRIMARY KEY,
domain TEXT UNIQUE,
first_seen TEXT,
last_updated TEXT
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS subdomains (
id INTEGER PRIMARY KEY,
domain_id INTEGER,
subdomain TEXT,
discovered_at TEXT,
FOREIGN KEY(domain_id) REFERENCES domains(id)
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS dns_records (
id INTEGER PRIMARY KEY,
domain_id INTEGER,
record_type TEXT,
record_value TEXT,
discovered_at TEXT,
FOREIGN KEY(domain_id) REFERENCES domains(id)
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS certificates (
id INTEGER PRIMARY KEY,
domain_id INTEGER,
name TEXT,
issuer TEXT,
valid_from TEXT,
valid_to TEXT,
discovered_at TEXT,
FOREIGN KEY(domain_id) REFERENCES domains(id)
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS shodan_ports (
id INTEGER PRIMARY KEY,
domain_id INTEGER,
ip TEXT,
port INTEGER,
data TEXT,
discovered_at TEXT,
FOREIGN KEY(domain_id) REFERENCES domains(id)
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS changes (
id INTEGER PRIMARY KEY,
domain TEXT,
change_type TEXT,
old_value TEXT,
new_value TEXT,
detected_at TEXT
)
''')
self.conn.commit()
def get_domain_id(self, domain):
cursor = self.conn.cursor()
cursor.execute("SELECT id FROM domains WHERE domain = ?", (domain,))
row = cursor.fetchone()
if row:
return row[0]
# Создаём новый
cursor.execute("INSERT INTO domains (domain, first_seen, last_updated) VALUES (?, ?, ?)",
(domain, datetime.now().isoformat(), datetime.now().isoformat()))
self.conn.commit()
return cursor.lastrowid
def save_subdomains(self, domain_id, subdomains):
cursor = self.conn.cursor()
for sub in subdomains:
cursor.execute("INSERT OR IGNORE INTO subdomains (domain_id, subdomain, discovered_at) VALUES (?, ?, ?)",
(domain_id, sub, datetime.now().isoformat()))
self.conn.commit()
def save_dns_records(self, domain_id, records):
cursor = self.conn.cursor()
for rtype, values in records.items():
for val in values:
cursor.execute("INSERT INTO dns_records (domain_id, record_type, record_value, discovered_at) VALUES (?, ?, ?, ?)",
(domain_id, rtype, val, datetime.now().isoformat()))
self.conn.commit()
def save_certificates(self, domain_id, certs):
cursor = self.conn.cursor()
for cert in certs:
cursor.execute("INSERT INTO certificates (domain_id, name, issuer, valid_from, valid_to, discovered_at) VALUES (?, ?, ?, ?, ?, ?)",
(domain_id, cert.get('name'), cert.get('issuer'), cert.get('valid_from'), cert.get('valid_to'), datetime.now().isoformat()))
self.conn.commit()
def log_change(self, domain, change_type, old_value, new_value):
cursor = self.conn.cursor()
cursor.execute("INSERT INTO changes (domain, change_type, old_value, new_value, detected_at) VALUES (?, ?, ?, ?, ?)",
(domain, change_type, old_value, new_value, datetime.now().isoformat()))
self.conn.commit()
def close(self):
self.conn.close()Этап 11: Главный конвейер (main.py)
#!/usr/bin/env python3
import os
import sys
import json
import yaml
from datetime import datetime
from utils.logger import setup_logger
from utils.db import OSINTDatabase
from utils.notifier import send_telegram
from modules.dns import get_dns_records
from modules.whois import get_whois_info
from modules.subdomains import get_all_subdomains
from modules.certificates import get_certificates_crtsh, get_certificate_expiry
from modules.shodan import get_shodan_host_by_domain
from modules.tech_stack import get_tech_wappalyzer
from modules.github import search_github_secrets
from modules.changes import compare_and_notify
# Загрузка конфига
with open('config/config.yaml', 'r') as f:
config = yaml.safe_load(f)
logger = setup_logger(config['logging']['level'], config['logging']['file'])
db = OSINTDatabase(config['database']['path'])
def process_domain(domain):
logger.info(f"Обработка {domain}")
# Получаем ID домена в БД
domain_id = db.get_domain_id(domain)
# 1. WHOIS
whois_data = get_whois_info(domain)
logger.info(f"WHOIS для {domain}: {whois_data.get('registrar', 'unknown')}")
# 2. DNS записи
dns = get_dns_records(domain)
db.save_dns_records(domain_id, dns)
logger.info(f"DNS: A={len(dns.get('A', []))}, MX={len(dns.get('MX', []))}")
# 3. Поддомены
sec_key = config['api_keys'].get('securitytrails')
subdomains = get_all_subdomains(domain, sec_key)
db.save_subdomains(domain_id, subdomains)
logger.info(f"Найдено {len(subdomains)} поддоменов")
# 4. Сертификаты
certs = get_certificates_crtsh(domain)
db.save_certificates(domain_id, certs)
expiring = get_certificate_expiry(certs)
if expiring:
logger.warning(f"Сертификаты истекают: {expiring}")
# 5. Shodan (по поддоменам, ограничимся первыми 10 IP)
shodan_key = config['api_keys'].get('shodan')
if shodan_key and subdomains:
for sub in subdomains[:10]:
hosts = get_shodan_host_by_domain(sub, shodan_key)
# сохраняем в БД
for h in hosts:
pass # упрощённо
# 6. Технологический стек
if subdomains:
main_url = f"https://{domain}"
tech = get_tech_wappalyzer(main_url)
logger.info(f"Технологии: {tech.get('technologies', [])}")
# 7. GitHub секреты
github_token = config['api_keys'].get('github')
if github_token:
secrets = search_github_secrets(domain, github_token)
if secrets:
logger.warning(f"Найдены потенциальные секреты в GitHub: {len(secrets)}")
send_telegram(
config['notifications']['telegram']['token'],
config['notifications']['telegram']['chat_id'],
f"Обнаружены секреты для {domain}: {len(secrets)} файлов"
)
# 8. Сравнение изменений
compare_and_notify(domain, config, db)
logger.info(f"Обработка {domain} завершена")
def main():
targets_file = config['targets']['file']
with open(targets_file, 'r') as f:
domains = [line.strip() for line in f if line.strip()]
for domain in domains:
try:
process_domain(domain)
except Exception as e:
logger.error(f"Ошибка для {domain}: {e}")
db.close()
logger.info("Все задачи выполнены")
if __name__ == "__main__":
main()Этап 12: Сравнение изменений и уведомления (modules/changes.py)
import json
import hashlib
import os
from utils.notifier import send_telegram
def load_previous_state(domain, data_dir='data/states'):
"""Загрузка предыдущего состояния из JSON"""
os.makedirs(data_dir, exist_ok=True)
state_file = os.path.join(data_dir, f"{domain}_state.json")
if os.path.exists(state_file):
with open(state_file, 'r') as f:
return json.load(f)
return {}
def save_current_state(domain, state, data_dir='data/states'):
state_file = os.path.join(data_dir, f"{domain}_state.json")
with open(state_file, 'w') as f:
json.dump(state, f, indent=2)
def compare_and_notify(domain, config, db):
"""Сравнение текущего состояния с предыдущим и уведомление об изменениях"""
# Собираем текущее состояние в словарь
current = {
'subdomains': sorted(db.get_subdomains(domain)),
'dns': db.get_dns_records(domain),
'certs': db.get_certificates(domain),
}
previous = load_previous_state(domain)
changes = []
# Сравнение поддоменов
if previous:
old_subs = set(previous.get('subdomains', []))
new_subs = set(current.get('subdomains', []))
added = new_subs - old_subs
removed = old_subs - new_subs
if added:
changes.append(f"Новые поддомены: {', '.join(added)}")
if removed:
changes.append(f"Удалены: {', '.join(removed)}")
# Если есть изменения — отправляем уведомление
if changes:
msg = f"Изменения для {domain}:\n" + "\n".join(changes)
send_telegram(
config['notifications']['telegram']['token'],
config['notifications']['telegram']['chat_id'],
msg
)
# Логируем в БД
for change in changes:
db.log_change(domain, 'change', '', change)
# Сохраняем новое состояние
save_current_state(domain, current)Этап 13: Утилита уведомлений (utils/notifier.py)
import requests
import smtplib
from email.mime.text import MIMEText
def send_telegram(token, chat_id, message):
"""Отправка уведомления в Telegram"""
if not token or not chat_id:
return
url = f"https://api.telegram.org/bot{token}/sendMessage"
try:
requests.post(url, json={'chat_id': chat_id, 'text': message}, timeout=10)
except Exception as e:
print(f"Telegram ошибка: {e}")
def send_email(smtp_server, port, username, password, to, subject, body):
"""Отправка email уведомления"""
msg = MIMEText(body)
msg['Subject'] = subject
msg['From'] = username
msg['To'] = to
try:
with smtplib.SMTP(smtp_server, port) as server:
server.starttls()
server.login(username, password)
server.send_message(msg)
except Exception as e:
print(f"Email ошибка: {e}")Этап 14: Bash-обёртка для cron (run.sh)
#!/bin/bash
# run.sh — обёртка для запуска конвейера с логированием
cd /path/to/osint-automation
# Активация виртуального окружения Python (если используется)
source venv/bin/activate
# Запуск основного скрипта
python main.py >> data/osint_cron.log 2>&1
# Проверка кода возврата
if [ $? -ne 0 ]; then
echo "Ошибка выполнения $(date)" >> data/osint_errors.log
fi
# Удаление старых логов (старше 30 дней)
find data/ -name "*.log" -mtime +30 -deleteНастройка cron (ежедневно в 8 утра):
0 8 * * * /path/to/osint-automation/run.sh
Этап 15: Обработка ошибок и rate limits
Важные аспекты при автоматизации:
- Ограничения API: Используйте задержки между запросами (time.sleep), храните время последнего запроса.
- Повторные попытки: Реализуйте повтор при временных ошибках (например, 429 Too Many Requests).
- Логирование: Все ошибки пишите в лог для последующего анализа.
- Таймауты: Все HTTP-запросы должны иметь таймаут, чтобы скрипт не зависал.
Пример декоратора для повторных попыток:
import time
import functools
def retry(max_attempts=3, delay=5):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts - 1:
raise
time.sleep(delay * (attempt + 1))
return None
return wrapper
return decoratorЭтап 16: Тестирование и мониторинг работы
После настройки автоматизации важно проверять:
- Работоспособность: Запустите скрипт вручную и проверьте вывод.
- Корректность данных: Проверьте, что база данных заполняется.
- Уведомления: Убедитесь, что при изменениях приходят сообщения.
- Логи: Проверьте файл логов на наличие ошибок.
- Используйте только публичные API и данные.
- Соблюдайте условия использования сервисов (Shodan, SecurityTrails, GitHub) — не превышайте лимиты запросов.
- Не пытайтесь обойти защиту (CAPTCHA, блокировки) нелегальными методами.
- Полученные данные используйте только для законных целей.
- При обнаружении уязвимостей или секретов сообщайте владельцам.
Автоматизация технического OSINT — это создание программного конвейера, который собирает, обрабатывает и сохраняет данные о цифровой инфраструктуре цели без участия человека. Модульная архитектура на Python с использованием готовых библиотек (dns.resolver, whois, requests, shodan) и внешних API (crt.sh, SecurityTrails, Shodan, GitHub) позволяет охватить все основные источники. Сравнение состояний и уведомления делают систему полезной для долгосрочного мониторинга. Регулярный запуск через cron превращает набор скриптов в постоянно работающую систему разведки.