#!/usr/bin/python # -*- coding: utf-8 -*- # --------------------------------------------------------------------------- # # # # Plugin for iSida Jabber Bot # # Copyright (c) diSabler # # Python 3 rewrite and modernized by OpenCode # # # # This program is free software: you can redistribute it and/or modify # # it under the terms of the GNU General Public License as published by # # the Free Software Foundation, either version 3 of the License, or # # (at your option) any later version. # # # # This program is distributed in the hope that it will be useful, # # but WITHOUT ANY WARRANTY; without even the implied warranty of # # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # # GNU General Public License for more details. # # # # You should have received a copy of the GNU General Public License # # along with this program. If not, see . # # # # --------------------------------------------------------------------------- # import os import re import json import subprocess import time from datetime import datetime # --------------------------------------------------------------------------- # # CONFIGURATION # # --------------------------------------------------------------------------- # CACHE_FILE = 'data/cache/domain_ng_cache.json' CACHE_TTL = 86400 # 24 hours WHOIS_TIMEOUT = 15 GEO_TIMEOUT = 10 # --------------------------------------------------------------------------- # # CACHE FUNCTIONS # # --------------------------------------------------------------------------- # def _get_cache_dir(): cache_dir = os.path.dirname(CACHE_FILE) if not os.path.exists(cache_dir): try: os.makedirs(cache_dir) except Exception: pass return cache_dir def _load_cache(): try: if os.path.exists(CACHE_FILE): with open(CACHE_FILE, 'r', encoding='utf-8') as f: data = json.load(f) now = time.time() return {k: v for k, v in data.items() if v.get('expires', 0) > now} except Exception: pass return {} def _save_cache(cache): try: _get_cache_dir() with open(CACHE_FILE, 'w', encoding='utf-8') as f: json.dump(cache, f, indent=2, ensure_ascii=False) except Exception: pass def _cache_get(key): cache = _load_cache() if key in cache: return cache[key].get('data') return None def _cache_set(key, data, ttl=CACHE_TTL): cache = _load_cache() cache[key] = { 'data': data, 'expires': time.time() + ttl } _save_cache(cache) # --------------------------------------------------------------------------- # # HELPER FUNCTIONS # # --------------------------------------------------------------------------- # def _is_valid_domain(text): text = text.strip().lower() if not text or len(text) < 3: return False if text.count('.') < 1: return False pattern = r'^[a-z0-9]([a-z0-9\-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9\-]*[a-z0-9])?)*$' return re.match(pattern, text) is not None def _is_valid_ip(text): text = text.strip() parts = text.split('.') if len(parts) != 4: return False try: for p in parts: num = int(p) if num < 0 or num > 255: return False return True except ValueError: return False def _is_valid_ip_or_domain(text): return _is_valid_domain(text) or _is_valid_ip(text) def _format_datetime(dt_str): if not dt_str or dt_str == 'N/A': return 'N/A' try: dt = datetime.fromisoformat(dt_str.replace('Z', '+00:00')) return dt.strftime('%Y-%m-%d %H:%M:%S UTC') except Exception: return dt_str def _run_command(cmd, timeout=15): try: proc = subprocess.run( cmd, capture_output=True, timeout=timeout, shell=True, encoding='utf-8', errors='replace' ) return proc.stdout, proc.stderr, proc.returncode except subprocess.TimeoutExpired: return '', 'Timeout', 124 except Exception as e: return '', str(e), 1 # --------------------------------------------------------------------------- # # CORE QUERY FUNCTIONS # # --------------------------------------------------------------------------- # def _whois_query(target, raw=False, brief=False): cache_key = f"whois:{target}:{raw}:{brief}" cached = _cache_get(cache_key) if cached is not None: return cached cmd = f"whois -H --no-recursion {target} 2>/dev/null" stdout, stderr, rc = _run_command(cmd, WHOIS_TIMEOUT) if rc != 0 or not stdout: result = {'success': False, 'error': 'No data found or whois command failed.'} _cache_set(cache_key, result) return result if raw: result = {'success': True, 'raw': stdout} _cache_set(cache_key, result) return result if brief: parsed = _parse_whois_brief(stdout, target) result = {'success': True, 'brief': parsed, 'raw': stdout} _cache_set(cache_key, result) return result parsed = _parse_whois_full(stdout, target) result = {'success': True, 'data': parsed, 'raw': stdout} _cache_set(cache_key, result) return result def _parse_whois_full(output, target): data = { 'domain': target, 'registrar': 'N/A', 'creation_date': 'N/A', 'expiry_date': 'N/A', 'updated_date': 'N/A', 'name_servers': [], 'status': [], 'dnssec': 'N/A', 'registrant': 'N/A' } lines = output.split('\n') for line in lines: line = line.strip() if not line: continue if '>>>' in line and not re.search(r'Last update', line, re.I): continue if re.search(r'^created:', line, re.I): val = line.split(':', 1)[1].strip() data['creation_date'] = _format_datetime(val) elif re.search(r'^paid-till:', line, re.I): val = line.split(':', 1)[1].strip() data['expiry_date'] = _format_datetime(val) elif re.search(r'Last updated|Last update', line, re.I): val = line.split(' on ', 1)[1].strip() if ' on ' in line else line.split(':', 1)[1].strip() data['updated_date'] = _format_datetime(val) elif re.search(r'^(Registrar:|Sponsoring Registrar:|registrar:)', line, re.I): data['registrar'] = line.split(':', 1)[1].strip() elif re.search(r'^(Creation Date:|Registered On:|Creation date:)', line, re.I): val = line.split(':', 1)[1].strip() data['creation_date'] = _format_datetime(val) elif re.search(r'^(Registry Expiry Date:|Expiration Date:|Expires On:)', line, re.I): val = line.split(':', 1)[1].strip() data['expiry_date'] = _format_datetime(val) elif re.search(r'^(Last updated:|Updated Date:|Last Updated On:)', line, re.I): val = re.sub(r'^(Last updated:|Updated Date:|Last Updated On:)', '', line, flags=re.I).strip() val = val.rstrip(' <<<').strip() data['updated_date'] = _format_datetime(val) elif re.search(r'^(nserver:|Name Server:)', line, re.I): ns = line.split(':', 1)[1].strip() if ns and ns not in data['name_servers']: data['name_servers'].append(ns) elif re.search(r'^(state:|Status:|Domain Status:)', line, re.I): status = line.split(':', 1)[1].strip() if status and status not in data['status']: data['status'].append(status) elif re.search(r'^DNSSEC:', line, re.I): data['dnssec'] = line.split(':', 1)[1].strip() elif re.search(r'^(person:|Registrant:|Registrant Name:)', line, re.I): data['registrant'] = line.split(':', 1)[1].strip() for k, v in list(data.items()): if isinstance(v, list) and not v: data[k] = 'N/A' elif isinstance(v, str) and not v.strip(): data[k] = 'N/A' elif isinstance(v, list): data[k] = ', '.join(v) if v else 'N/A' return data def _parse_whois_brief(output, target): data = { 'domain': target, 'registrar': 'N/A', 'creation_date': 'N/A', 'expiry_date': 'N/A', 'name_servers': 'N/A' } lines = output.split('\n') ns_list = [] for line in lines: line = line.strip() if not line: continue if '>>>' in line and not re.search(r'Last update', line, re.I): continue if re.search(r'^created:', line, re.I): val = line.split(':', 1)[1].strip() data['creation_date'] = _format_datetime(val) elif re.search(r'^paid-till:', line, re.I): val = line.split(':', 1)[1].strip() data['expiry_date'] = _format_datetime(val) elif re.search(r'^(Registrar:|Sponsoring Registrar:|registrar:)', line, re.I): data['registrar'] = line.split(':', 1)[1].strip() elif re.search(r'^(Creation Date:|Registered On:|Creation date:)', line, re.I): val = line.split(':', 1)[1].strip() data['creation_date'] = _format_datetime(val) elif re.search(r'^(Registry Expiry Date:|Expiration Date:|Expires On:)', line, re.I): val = line.split(':', 1)[1].strip() data['expiry_date'] = _format_datetime(val) elif re.search(r'^(nserver:|Name Server:)', line, re.I): ns = line.split(':', 1)[1].strip() if ns and ns not in ns_list: ns_list.append(ns) data['name_servers'] = ', '.join(ns_list) if ns_list else 'N/A' return data def _geo_ip(ip): cache_key = f"geo:{ip}" cached = _cache_get(cache_key) if cached is not None: return cached cmd = f"curl -s -m {GEO_TIMEOUT} 'http://ip-api.com/json/{ip}' 2>/dev/null" stdout, stderr, rc = _run_command(cmd, GEO_TIMEOUT) if rc != 0 or not stdout: result = {'success': False, 'error': 'Failed to fetch geolocation data.'} _cache_set(cache_key, result, 3600) return result try: data = json.loads(stdout) if data.get('status') != 'success': result = {'success': False, 'error': 'No geolocation data available.'} _cache_set(cache_key, result, 3600) return result result = { 'success': True, 'country': data.get('country', 'N/A'), 'country_code': data.get('countryCode', 'N/A'), 'region': data.get('regionName', 'N/A'), 'city': data.get('city', 'N/A'), 'isp': data.get('isp', 'N/A'), 'org': data.get('org', 'N/A'), 'lat': data.get('lat', 0), 'lon': data.get('lon', 0), 'timezone': data.get('timezone', 'N/A') } _cache_set(cache_key, result, 3600) return result except json.JSONDecodeError: result = {'success': False, 'error': 'Invalid response from geolocation service.'} _cache_set(cache_key, result, 3600) return result # --------------------------------------------------------------------------- # # COMMAND HANDLERS # # --------------------------------------------------------------------------- # def cmd_domain(type, jid, nick, text): text = text.strip() if not text: send_msg(type, jid, nick, 'Usage: d [raw|brief|geo]\n' 'Examples:\n' ' d example.com\n' ' d example.com raw\n' ' d example.com brief\n' ' d 8.8.8.8 geo' ) return parts = text.split() target = parts[0] subcmd = parts[1] if len(parts) > 1 else '' if subcmd.lower() not in ['raw', 'brief', 'geo'] and len(parts) > 1: send_msg(type, jid, nick, f'Invalid subcommand: {subcmd}. Use: raw, brief, geo') return if not _is_valid_ip_or_domain(target): send_msg(type, jid, nick, f'Invalid domain or IP address: {target}') return if subcmd.lower() == 'raw': result = _whois_query(target, raw=True) if result.get('success'): raw_output = result.get('raw', '') if len(raw_output) > 15000: raw_output = raw_output[:15000] + '\n... (truncated)' send_msg(type, jid, nick, f'WHOIS RAW for {target}:\n{raw_output}') else: send_msg(type, jid, nick, f'[X] {result.get("error", "Unknown error")}') return if subcmd.lower() == 'brief': result = _whois_query(target, brief=True) if result.get('success'): data = result.get('brief', {}) msg = f"WHOIS (brief) for {data.get('domain', target)}\n" msg += f"Registrar: {data.get('registrar', 'N/A')}\n" msg += f"Created: {data.get('creation_date', 'N/A')}\n" msg += f"Expires: {data.get('expiry_date', 'N/A')}\n" msg += f"Name Servers: {data.get('name_servers', 'N/A')}" send_msg(type, jid, nick, msg) else: send_msg(type, jid, nick, f'[X] {result.get("error", "Unknown error")}') return if subcmd.lower() == 'geo': if not _is_valid_ip(target): cmd = f"dig +short {target} | head -1 2>/dev/null" stdout, _, rc = _run_command(cmd, 5) if rc == 0 and stdout.strip(): target_ip = stdout.strip() if _is_valid_ip(target_ip): target = target_ip else: send_msg(type, jid, nick, f'[X] Could not resolve {target} to an IP address.') return else: send_msg(type, jid, nick, f'[X] Could not resolve {target} to an IP address.') return result = _geo_ip(target) if result.get('success'): msg = f"Geolocation for {target}\n" msg += f"Country: {result.get('country', 'N/A')} ({result.get('country_code', 'N/A')})\n" msg += f"City: {result.get('city', 'N/A')}\n" msg += f"Region: {result.get('region', 'N/A')}\n" msg += f"ISP: {result.get('isp', 'N/A')}\n" msg += f"Org: {result.get('org', 'N/A')}\n" msg += f"Coordinates: {result.get('lat', 0):.4f}, {result.get('lon', 0):.4f}\n" msg += f"Timezone: {result.get('timezone', 'N/A')}" send_msg(type, jid, nick, msg) else: send_msg(type, jid, nick, f'[X] {result.get("error", "Unknown error")}') return # Default: full whois result = _whois_query(target) if result.get('success'): data = result.get('data', {}) msg = f"WHOIS for {data.get('domain', target)}\n" msg += f"Registrar: {data.get('registrar', 'N/A')}\n" msg += f"Created: {data.get('creation_date', 'N/A')}\n" msg += f"Expires: {data.get('expiry_date', 'N/A')}\n" msg += f"Updated: {data.get('updated_date', 'N/A')}\n" msg += f"Name Servers: {data.get('name_servers', 'N/A')}\n" msg += f"Status: {data.get('status', 'N/A')}\n" msg += f"DNSSEC: {data.get('dnssec', 'N/A')}\n" if data.get('registrant', 'N/A') != 'N/A': msg += f"Registrant: {data.get('registrant')}" send_msg(type, jid, nick, msg) else: send_msg(type, jid, nick, f'[X] {result.get("error", "Unknown error")}') # --------------------------------------------------------------------------- # # ALIAS FOR OLD COMMANDS # # --------------------------------------------------------------------------- # def cmd_domain_info(type, jid, nick, text): cmd_domain(type, jid, nick, text) def cmd_domain_info_raw(type, jid, nick, text): cmd_domain(type, jid, nick, f"{text} raw") # --------------------------------------------------------------------------- # # REGISTER COMMANDS # # --------------------------------------------------------------------------- # execute = [ (3, 'd', cmd_domain, 2, 'Domain info: d [raw|brief|geo]'), (3, 'domain_info', cmd_domain_info, 2, 'Alias for d (legacy)'), (3, 'domain_info_raw', cmd_domain_info_raw, 2, 'Alias for d raw (legacy)'), ]