189 lines
7.6 KiB
Python
189 lines
7.6 KiB
Python
"""
|
||
This plugin provides IP geolocation functionality using free APIs.
|
||
It uses ip-api.com as the primary API with a fallback to ipapi.co.
|
||
"""
|
||
|
||
import logging
|
||
import aiohttp
|
||
import simplematrixbotlib as botlib
|
||
import socket
|
||
import re
|
||
|
||
from plugins.utils import is_public_destination
|
||
|
||
async def is_valid_ip(ip):
|
||
"""Check if the provided string is a valid IP address."""
|
||
try:
|
||
socket.inet_pton(socket.AF_INET, ip)
|
||
return True
|
||
except socket.error:
|
||
try:
|
||
socket.inet_pton(socket.AF_INET6, ip)
|
||
return True
|
||
except socket.error:
|
||
return False
|
||
|
||
def is_domain(domain):
|
||
"""Check if the provided string is a domain name."""
|
||
domain_pattern = re.compile(
|
||
r'^([a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}$'
|
||
)
|
||
return bool(domain_pattern.match(domain))
|
||
|
||
async def resolve_domain(domain):
|
||
"""Resolve a domain name to an IP address."""
|
||
try:
|
||
return socket.gethostbyname(domain)
|
||
except socket.gaierror:
|
||
return None
|
||
|
||
async def query_ip_api_com(ip):
|
||
"""Query ip-api.com for geolocation information."""
|
||
url = f"http://ip-api.com/json/{ip}"
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.get(url) as response:
|
||
if response.status == 200:
|
||
data = await response.json()
|
||
return data
|
||
else:
|
||
logging.error(f"ip-api.com returned status {response.status}")
|
||
return None
|
||
except Exception as e:
|
||
logging.error(f"Error querying ip-api.com: {e}")
|
||
return None
|
||
|
||
async def query_ipapi_co(ip):
|
||
"""Query ipapi.co for geolocation information (fallback)."""
|
||
url = f"https://ipapi.co/{ip}/json/"
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.get(url) as response:
|
||
if response.status == 200:
|
||
data = await response.json()
|
||
return data
|
||
else:
|
||
logging.error(f"ipapi.co returned status {response.status}")
|
||
return None
|
||
except Exception as e:
|
||
logging.error(f"Error querying ipapi.co: {e}")
|
||
return None
|
||
|
||
async def query_geolocation(ip):
|
||
"""Query geolocation information using primary and fallback APIs."""
|
||
data = await query_ip_api_com(ip)
|
||
if not data or data.get('status') == 'fail':
|
||
logging.info("Primary API failed, trying fallback API")
|
||
data = await query_ipapi_co(ip)
|
||
return data
|
||
|
||
async def format_geolocation_results(ip, data):
|
||
"""Format geolocation results into a readable message."""
|
||
if not data:
|
||
return f"🔍 No geolocation data found for {ip}."
|
||
if 'status' in data and data.get('status') == 'fail':
|
||
return f"🔍 No geolocation data found for {ip}."
|
||
if 'country' in data:
|
||
country = data.get('country', 'N/A')
|
||
country_code = data.get('countryCode', 'N/A')
|
||
region = data.get('regionName', data.get('region', 'N/A'))
|
||
city = data.get('city', 'N/A')
|
||
postal = data.get('zip', 'N/A')
|
||
latitude = data.get('lat', 'N/A')
|
||
longitude = data.get('lon', 'N/A')
|
||
timezone = data.get('timezone', 'N/A')
|
||
isp = data.get('isp', 'N/A')
|
||
org = data.get('org', 'N/A')
|
||
asn = data.get('as', 'N/A')
|
||
else:
|
||
country = data.get('country_name', data.get('country', 'N/A'))
|
||
country_code = data.get('country_code', data.get('countryCode', 'N/A'))
|
||
region = data.get('region', 'N/A')
|
||
city = data.get('city', 'N/A')
|
||
postal = data.get('postal', 'N/A')
|
||
latitude = data.get('latitude', 'N/A')
|
||
longitude = data.get('longitude', 'N/A')
|
||
timezone = data.get('timezone', 'N/A')
|
||
isp = data.get('org', 'N/A')
|
||
org = data.get('org', 'N/A')
|
||
asn = data.get('asn', 'N/A')
|
||
content = f"<strong>🔍 IP Geolocation Results for {ip}</strong><br><br>"
|
||
content += f"<strong>Country:</strong> {country} ({country_code})<br>"
|
||
content += f"<strong>Region:</strong> {region}<br>"
|
||
content += f"<strong>City:</strong> {city}<br>"
|
||
content += f"<strong>Postal Code:</strong> {postal}<br>"
|
||
content += f"<strong>Coordinates:</strong> {latitude}, {longitude}<br>"
|
||
content += f"<strong>Timezone:</strong> {timezone}<br>"
|
||
content += f"<strong>ISP/Organization:</strong> {isp}<br>"
|
||
content += f"<strong>ASN:</strong> {asn}<br>"
|
||
message = f"<details><summary><strong>🔍 Geolocation: {ip}</strong></summary>{content}</details>"
|
||
return message
|
||
|
||
async def handle_command(room, message, bot, prefix, config):
|
||
"""Handle the !geo command."""
|
||
match = botlib.MessageMatch(room, message, bot, prefix)
|
||
if match.is_not_from_this_bot() and match.prefix() and match.command("geo"):
|
||
args = match.args()
|
||
if len(args) < 1:
|
||
await bot.api.send_text_message(
|
||
room.room_id,
|
||
"Usage: !geo <ip_address/domain>\nExample: !geo 8.8.8.8\nExample: !geo example.com"
|
||
)
|
||
return
|
||
query = args[0].strip()
|
||
logging.info(f"Received !geo command for: {query}")
|
||
try:
|
||
ip = query
|
||
if is_domain(query):
|
||
await bot.api.send_text_message(
|
||
room.room_id,
|
||
f"🔍 Resolving domain {query} to IP address..."
|
||
)
|
||
ip = await resolve_domain(query)
|
||
if not ip:
|
||
await bot.api.send_text_message(room.room_id,
|
||
f"Failed to resolve domain {query} to IP address.")
|
||
return
|
||
if not is_public_destination(ip):
|
||
await bot.api.send_text_message(room.room_id,
|
||
"❌ That domain resolves to a private/internal IP, geo not allowed.")
|
||
return
|
||
await bot.api.send_text_message(room.room_id,
|
||
f"Domain {query} resolved to IP {ip}")
|
||
elif not await is_valid_ip(query):
|
||
await bot.api.send_text_message(room.room_id,
|
||
f"Invalid IP address or domain format: {query}")
|
||
return
|
||
else:
|
||
if not is_public_destination(ip):
|
||
await bot.api.send_text_message(room.room_id,
|
||
"❌ Geolocation of private IP addresses is not allowed.")
|
||
return
|
||
await bot.api.send_text_message(room.room_id,
|
||
f"🔍 Looking up geolocation for {ip}...")
|
||
geo_data = await query_geolocation(ip)
|
||
result_message = await format_geolocation_results(ip, geo_data)
|
||
await bot.api.send_markdown_message(room.room_id, result_message)
|
||
logging.info(f"Successfully sent geolocation results for {ip}")
|
||
except Exception as e:
|
||
await bot.api.send_text_message(room.room_id,
|
||
f"An error occurred during geolocation lookup for {query}. Please try again later.")
|
||
logging.error(f"Error in geo plugin for {query}: {e}", exc_info=True)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Plugin Metadata
|
||
# ---------------------------------------------------------------------------
|
||
__version__ = "1.0.1"
|
||
__author__ = "Funguy Bot"
|
||
__description__ = "IP geolocation lookup"
|
||
__help__ = """
|
||
<details>
|
||
<summary><strong>!geo</strong> – IP / domain geolocation</summary>
|
||
<ul>
|
||
<li><code>!geo <ip></code> – Locate an IP address</li>
|
||
<li><code>!geo <domain></code> – Resolves domain then locates</li>
|
||
</ul>
|
||
<p>Shows country, region, city, coordinates, ISP, ASN. Uses ip-api.com / ipapi.co.</p>
|
||
</details>
|
||
"""
|