Этот Python-скрипт при каждом запуске проверяет список отслеживаемых IP-адресов по признаку threat-intelligence бесплатного API ip.xakki.pro, сравнивает результат с локальным JSON-файлом состояния от предыдущего запуска и выводит строку ALERT в момент перехода IP из чистого в помеченный как вредоносный — предназначен для периодического запуска через cron, а не для самостоятельного цикла внутри скрипта.
Зависимости: нет — только стандартная библиотека Python 3.8+ (urllib, json, csv, ipaddress, argparse). pip install не требуется.
#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""threat-monitor.py -- alert when a watched IP transitions from clean to
flagged in the threat-intelligence feeds behind ip.xakki.pro.
Meant to be invoked periodically (see the cron example below), not run in a
loop itself: each invocation checks every IP in --ip-file once via
GET /api/v1/ip?ip=<addr>&include=security, compares threat.isMalicious
against the last known state in a local JSON state file, and prints an
ALERT line (to stdout) the moment an IP flips false -> true (or is seen
flagged for the first time). A clean-to-clean or already-flagged IP prints
nothing, so cron output stays quiet except on a real transition.
This is a genuinely repeating, unattended use of the free anonymous tier --
the epic that authored this section considered whether that competes with
the paid monitoring product and decided, deliberately, to keep this recipe
as-is (see the card's Decisions). It works entirely within the same
per-unique-IP/day metering every other recipe in this section respects.
Zero dependencies -- stdlib only (urllib, json, argparse; no smtplib wiring
by default, to keep the zero-dependency, zero-config bar -- see "Extending
this script" in the comment near the bottom for where to add an email/
webhook hook without adding a package).
Usage:
python3 threat-monitor.py --ip-file watched-ips.txt --state-file state.json
Example crontab entry (checks every 15 minutes, appends to a log):
*/15 * * * * /usr/bin/python3 /path/to/threat-monitor.py \\
--ip-file /path/to/watched-ips.txt \\
--state-file /path/to/state.json >> /path/to/threat-monitor.log 2>&1
"""
import argparse
import ipaddress
import json
import os
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
DEFAULT_BASE_URL = "https://ip.xakki.pro"
USER_AGENT = "myip-threat-monitor-recipe/1.0 (+https://ip.xakki.pro/tools/scripts/threat-monitor.html)"
def load_state(path):
if not os.path.exists(path):
return {}
with open(path, encoding="utf-8") as f:
return json.load(f)
def save_state(path, state):
tmp_path = path + ".tmp"
with open(tmp_path, "w", encoding="utf-8") as f:
json.dump(state, f, indent=2, sort_keys=True)
os.replace(tmp_path, path) # atomic on POSIX and Windows alike
def fetch_security(base_url, ip_value, max_retries=5, base_backoff=0.5):
url = "{}/api/v1/ip?ip={}&include=security".format(base_url, ip_value)
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "application/json"})
attempt = 0
while True:
try:
with urllib.request.urlopen(req, timeout=15) as resp:
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
if not (e.code == 429 or e.code >= 500) or attempt >= max_retries:
raise
wait = float(e.headers.get("Retry-After", 0)) or (base_backoff * (2 ** attempt))
except (urllib.error.URLError, TimeoutError, ConnectionError):
if attempt >= max_retries:
raise
wait = base_backoff * (2 ** attempt)
attempt += 1
time.sleep(wait)
def main():
parser = argparse.ArgumentParser(description="Alert when a watched IP starts showing up in threat feeds.")
parser.add_argument("--ip-file", required=True, help="Text file, one IP address per line, to watch.")
parser.add_argument("--state-file", required=True, help="JSON file this script reads/writes between runs.")
parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help="API base URL (default: production).")
parser.add_argument("--delay", type=float, default=0.5, help="Seconds between lookups (default: 0.5).")
args = parser.parse_args()
with open(args.ip_file, encoding="utf-8") as f:
watched_ips = [line.strip() for line in f if line.strip()]
state = load_state(args.state_file)
now = datetime.now(timezone.utc).isoformat()
alerts = 0
for i, ip_value in enumerate(watched_ips):
try:
ipaddress.ip_address(ip_value)
except ValueError:
print("skipping invalid IP in --ip-file: {!r}".format(ip_value), file=sys.stderr)
continue
try:
data = fetch_security(args.base_url, ip_value)
except Exception as e: # noqa: BLE001
print("{} {} -- lookup failed, leaving last known state untouched: {}".format(now, ip_value, e),
file=sys.stderr)
if i < len(watched_ips) - 1:
time.sleep(args.delay)
continue
if "security" in (data.get("limited") or []):
# Withheld this run -- NOT the same as "verified clean". Leave
# the stored state exactly as it was rather than overwriting a
# real prior verdict with "no data", and never treat a withheld
# check as a silent false->false (which could mask a real
# transition that simply couldn't be checked this run).
print("{} {} -- security group rate-limited this run, skipping".format(now, ip_value), file=sys.stderr)
if i < len(watched_ips) - 1:
time.sleep(args.delay)
continue
threat = data.get("threat") or {}
is_malicious = bool(threat.get("isMalicious"))
previous = state.get(ip_value, {}).get("is_malicious", False)
if is_malicious and not previous:
sources = ", ".join(threat.get("sources") or []) or "unspecified feed"
print("{} ALERT: {} is now flagged malicious (sources: {})".format(now, ip_value, sources))
alerts += 1
elif not is_malicious and previous:
print("{} INFO: {} is no longer flagged malicious".format(now, ip_value))
state[ip_value] = {"is_malicious": is_malicious, "checked_at": now}
if i < len(watched_ips) - 1:
time.sleep(args.delay)
save_state(args.state_file, state)
if alerts == 0:
print("{} no new alerts ({} IP(s) checked)".format(now, len(watched_ips)), file=sys.stderr)
# Extending this script: to email or webhook-notify instead of just printing,
# wrap the two print(...) calls tagged "ALERT"/"INFO" above with your own
# notify(message) function -- e.g. smtplib.SMTP (stdlib, still zero
# dependencies) for email, or urllib.request.urlopen with a JSON POST body
# for a Slack/Discord/Telegram webhook. Left out of this recipe on purpose:
# every reader's SMTP/webhook credentials are different, and a script that
# fails silently because of an unset environment variable is worse than a
# script that just prints and lets cron's own mail-on-output handle delivery.
if __name__ == "__main__":
main()
Скрипт распространяется по лицензии MIT (заголовок SPDX-License-Identifier: MIT в начале файла) — копируйте, изменяйте и используйте свободно в своём проекте, указывать авторство не обязательно.
Именно сохранение состояния делает этот скрипт МОНИТОРОМ, а не разовой проверкой: threat.isMalicious для каждого отслеживаемого IP записывается в локальный JSON-файл после каждого запуска, а следующий запуск сравнивает с ним, чтобы обнаружить настоящий ПЕРЕХОД из false в true, и только тогда выводит ALERT — а не при каждом запуске, когда IP уже помечен, что завалило бы почтовый ящик cron одним и тем же оповещением каждые 15 минут бесконечно. Запись состояния атомарна (запись во временный файл .tmp, затем os.replace), поэтому сбой в середине прогона никогда не портит файл состояния до нечитаемого частичного JSON.
Ответ с ограничением частоты (группа «security» придержана) для конкретного IP оставляет сохранённое состояние этого IP ПОЛНОСТЬЮ НЕТРОНУТЫМ в данном прогоне, а не перезаписывает настоящий предыдущий вердикт значением «нет данных» и не молча фиксирует false->false. Перезапись рисковала бы замаскировать настоящий переход, который просто не удалось проверить в этом цикле — следующий успешный прогон продолжит ровно с того места, где остановилось последнее подтверждённое состояние.
По умолчанию не подключены smtplib/webhook — у каждого читателя свои SMTP-реквизиты или URL webhook, а скрипт мониторинга, который тихо ломается из-за одной незаданной переменной окружения, хуже скрипта, который просто печатает в stdout/stderr и отдаёт доставку встроенному в cron mail-on-output (или вашей собственной обёртке). Комментарий «Extending this script» ближе к концу файла показывает, куда именно добавить такой хук, используя только стандартную библиотеку (smtplib сам по себе входит в stdlib — рецепт в любом случае остаётся без внешних зависимостей).
GET /api/v1/ip никогда не отвечает жёстким 429 от самого приложения — группы обогащения (geo, network, security) лимитируются по уникальному целевому IP в сутки (UTC) плюс более жёсткий поминутный бюджет; при превышении группа мягко придерживается: HTTP 200, "status":"partial", имя группы в "limited", и заголовки RateLimit-Limit/RateLimit-Remaining/RateLimit-Reset в ответе. Этот скрипт читает эти сигналы, а не угадывает, и явно фиксирует придержанный результат (никогда как пустое/null значение — это выглядело бы так же, как «у API нет данных по этому IP»). Настоящий 429/503 всё же возможен на границе — это собственный анти-флуд nginx, независимый от лимитов выше — если слать запросы слишком быстро с одной машины; скрипт делает экспоненциальные повторы на 429/5xx и учитывает заголовок Retry-After, если сервер его прислал. У анонимных вызовов небольшой суточный/поминутный бюджет на группу (см. заголовок RateLimit-Limit в любом ответе или /api/v1/tiers); для заметно бо́льшего бюджета зарегистрируйте бесплатный API-токен.
Сам этот скрипт и есть пример работы с API — отдельный блок curl/ответ не нужен, код выше и есть запрос. Полная схема ответа и все параметры — в документации API:
Понравился инструмент? Бесплатный аккаунт нашего API даёт 1000 запросов в день.Смотреть документацию API →