This Python script checks a list of watched IP addresses against the free ip.xakki.pro API's threat-intelligence signal each time it runs, compares the result to a local JSON state file from the previous run, and prints an ALERT line the moment an IP transitions from clean to flagged malicious — meant to be invoked periodically by cron, not run in a loop itself.
placeholder
#!/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()
This script is MIT-licensed (SPDX-License-Identifier: MIT header at the top of the file) — copy, modify and reuse it freely in your own project, no attribution required.
State persistence is what makes this a MONITOR rather than a one-off check: threat.isMalicious for each watched IP is written to a local JSON file after every run, and the next run compares against it to detect a genuine false-to-true TRANSITION, printing an ALERT only then — not on every run an IP happens to already be flagged, which would flood a cron mailbox with the same alert every 15 minutes forever. The state write is atomic (write to a .tmp file, then os.replace) so a crash mid-run never corrupts the state file into unreadable partial JSON.
A rate-limited ("security" group withheld) response for a given IP leaves that IP's stored state COMPLETELY UNTOUCHED for this run, rather than overwriting a real prior verdict with "no data" or silently recording a false->false. Overwriting it would risk masking a genuine transition that simply couldn't be checked this cycle — the next successful run picks up exactly where the last verified state left off.
No smtplib/webhook wiring ships by default — every reader's SMTP credentials or webhook URL are different, and a monitoring script that fails silently because of one unset environment variable is worse than one that just prints to stdout/stderr and lets cron's own mail-on-output (or your own wrapper) handle delivery. The "Extending this script" comment near the bottom of the file shows exactly where to add that hook using only stdlib (smtplib is itself stdlib — this stays a zero-dependency recipe either way).
GET /api/v1/ip never returns a hard 429 from the app itself — its enrichment groups (geo, network, security) are metered per unique target IP per UTC day, plus a tighter per-minute budget, and going over either soft-withholds the group instead: HTTP 200, "status":"partial", the group's name in "limited", and RateLimit-Limit/RateLimit-Remaining/RateLimit-Reset headers on the response. This script reads those signals rather than guessing, and records a withheld result explicitly (never as a blank/null, which would look identical to "the API has no data for this IP"). A real 429/503 can still happen at the edge — nginx's own flood guard, independent of the metering above — if you call the endpoint too fast from one machine; the script backs off exponentially on any 429/5xx and honors a Retry-After header when the server sends one. Anonymous callers get a modest daily/per-minute budget per group (see the RateLimit-Limit header on any response, or /api/v1/tiers); register a free API token for a much larger one if you plan to run this against more than a handful of IPs.
This script itself is the API example — there is no separate curl/response block to show, because the code above already IS the request. See the full response schema and every parameter in the API docs:
Like this tool? Our API gives free accounts 1,000 requests/day.See the API docs →