23b4d6f2a4
GeoIP and ASN lookup via MaxMind GeoLite2 mmdb, Tor exit node check against local bulk exit list, IP reputation via Firehol/ET blocklist feeds, and CVE lookup against local NVD JSON mirror. Includes cron-friendly update script (scripts/update-data.sh) for all data sources and make update-data target. GeoLite2 requires a free MaxMind license key; all other sources are freely downloadable. Plugins: geoip, asn, torcheck, iprep, cve Commands: !geoip, !asn, !tor, !iprep, !cve Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
99 lines
2.6 KiB
Python
99 lines
2.6 KiB
Python
"""Plugin: GeoIP lookup using MaxMind GeoLite2-City mmdb."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
from derp.plugin import command
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_DB_PATHS = [
|
|
Path("data/GeoLite2-City.mmdb"),
|
|
Path("/usr/share/GeoIP/GeoLite2-City.mmdb"),
|
|
Path.home() / ".local" / "share" / "GeoIP" / "GeoLite2-City.mmdb",
|
|
]
|
|
|
|
_reader = None
|
|
|
|
|
|
def _get_reader():
|
|
"""Lazy-load the mmdb reader."""
|
|
global _reader
|
|
if _reader is not None:
|
|
return _reader
|
|
try:
|
|
import maxminddb
|
|
except ImportError:
|
|
log.error("maxminddb package not installed")
|
|
return None
|
|
for path in _DB_PATHS:
|
|
if path.is_file():
|
|
_reader = maxminddb.open_database(str(path))
|
|
log.info("geoip: loaded %s", path)
|
|
return _reader
|
|
log.warning("geoip: no GeoLite2-City.mmdb found")
|
|
return None
|
|
|
|
|
|
@command("geoip", help="GeoIP lookup: !geoip <ip>")
|
|
async def cmd_geoip(bot, message):
|
|
"""Look up geographic location for an IP address.
|
|
|
|
Usage:
|
|
!geoip 8.8.8.8
|
|
"""
|
|
parts = message.text.split(None, 2)
|
|
if len(parts) < 2:
|
|
await bot.reply(message, "Usage: !geoip <ip>")
|
|
return
|
|
|
|
addr = parts[1]
|
|
try:
|
|
ip = ipaddress.ip_address(addr)
|
|
except ValueError:
|
|
await bot.reply(message, f"Invalid IP address: {addr}")
|
|
return
|
|
|
|
if ip.is_private or ip.is_loopback:
|
|
await bot.reply(message, f"{addr}: private/loopback address")
|
|
return
|
|
|
|
reader = _get_reader()
|
|
if reader is None:
|
|
await bot.reply(message, "GeoIP database not available (run update-data)")
|
|
return
|
|
|
|
try:
|
|
rec = reader.get(str(ip))
|
|
except Exception as exc:
|
|
await bot.reply(message, f"Lookup error: {exc}")
|
|
return
|
|
|
|
if not rec:
|
|
await bot.reply(message, f"{addr}: no GeoIP data")
|
|
return
|
|
|
|
country = rec.get("country", {}).get("names", {}).get("en", "")
|
|
iso = rec.get("country", {}).get("iso_code", "")
|
|
city = rec.get("city", {}).get("names", {}).get("en", "")
|
|
loc = rec.get("location", {})
|
|
lat = loc.get("latitude")
|
|
lon = loc.get("longitude")
|
|
tz = loc.get("time_zone", "")
|
|
|
|
info = []
|
|
if city and country:
|
|
info.append(f"{city}, {country} ({iso})")
|
|
elif country:
|
|
info.append(f"{country} ({iso})")
|
|
if lat is not None and lon is not None:
|
|
info.append(f"{lat:.4f}, {lon:.4f}")
|
|
if tz:
|
|
info.append(tz)
|
|
|
|
result = f"{addr}: {' | '.join(info)}" if info else f"{addr}: no location data"
|
|
await bot.reply(message, result)
|