#!/usr/bin/python # -*- coding: utf-8 -*- # --------------------------------------------------------------------------- # # # # Plugin for iSida Jabber Bot # # Copyright (C) diSabler # # # # 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 subprocess import socket # --------------------------------------------------------------------------- # # 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 # --------------------------------------------------------------------------- # # SHELL EXECUTE WRAPPER # # --------------------------------------------------------------------------- # def _shell_execute(cmd, rn=''): """ Execute a shell command and return its output as a string. Uses subprocess.run() with text=True for proper Python 3 string handling. """ if GT('paranoia_mode'): return L('Command temporary blocked!', rn) try: result = subprocess.run( cmd, shell=True, capture_output=True, text=True, timeout=15 ) if result.stdout and result.stdout.strip(): return remove_sub_space(result.stdout.strip()) elif result.stderr and result.stderr.strip(): return remove_sub_space(result.stderr.strip()) elif result.returncode == 0: return L('ok', rn) else: return L('Command execution error.', rn) except subprocess.TimeoutExpired: return L('Timeout exceeded.', rn) except FileNotFoundError: return L('Command not found.', rn) except Exception as e: return L('I can\'t execute it! Error: %s', rn) % str(e) # --------------------------------------------------------------------------- # # NETWORK COMMANDS # # --------------------------------------------------------------------------- # def net_ping(type, jid, nick, text): text = text.strip().lower() if '.' in text and len(text) > 4 and re.match(r'[-0-9a-z.]+\Z', text, re.U+re.I): msg = deidna(_shell_execute('ping -c4 %s' % text, '%s/%s' % (jid, nick))) else: msg = L('Smoke help about command!', '%s/%s' % (jid, nick)) send_msg(type, jid, nick, msg) def get_tld(type, jid, nick, text): if len(text) >= 2: tld = readfile(tld_list) tld = tld.split('\n') msg = L('Not found!', '%s/%s' % (jid, nick)) for tl in tld: if tl.split('\t')[0].lower() == text.lower(): msg = '.' + tl.replace('\t', ' - ', 1).replace('\t', '\n') break else: msg = L('What do you want to find?', '%s/%s' % (jid, nick)) send_msg(type, jid, nick, msg) def get_dns(type, jid, nick, text): if _is_valid_ip(text): try: msg = socket.gethostbyaddr(text)[0] except: msg = L('I can\'t resolve it', '%s/%s' % (jid, nick)) else: try: ans = socket.getaddrinfo(text, None, 0, 0, socket.IPPROTO_TCP) msg = text + ' - ' for an in ans: msg += an[4][0] + ' | ' msg = msg[:-2] except: msg = L('I can\'t resolve it', '%s/%s' % (jid, nick)) send_msg(type, jid, nick, msg) def srv_nslookup(type, jid, nick, text): srv_raw_check(type, jid, nick, 'nslookup ' + text) def srv_dig(type, jid, nick, text): srv_raw_check(type, jid, nick, 'dig ' + text) def srv_host(type, jid, nick, text): srv_raw_check(type, jid, nick, 'host ' + text) def srv_raw_check(type, jid, nick, text): text = enidna_raw(text) text = ''.join(re.findall(u'[-a-z0-9\.\_\?\#\=\@\%\ \+]+', text, re.S | re.I)[0]) send_msg(type, jid, nick, deidna(_shell_execute(text, '%s/%s' % (jid, nick)))) def chkserver(type, jid, nick, text): for a in ':;&/|\\\n\t\r': text = text.replace(a, ' ') t = re.findall(u'[-a-zа-я0-9._?#=@%]+', text, re.S | re.I | re.U) if len(t) >= 2: port = [] for a in t: if a.isdigit(): port.append(a) for a in port: t.remove(a) if len(t) == 1 and len(port) >= 1: t = t[0] port.sort() msg = _shell_execute('nmap %s -p%s -P0 -T5' % (t, ','.join(port)), '%s/%s' % (jid, nick)) try: msg = '%s\n%s' % (t, reduce_spaces_all(re.findall('SERVICE(.*)Nmap', msg, re.S | re.U)[0][1:-2])) except: try: msg = '' for a in port: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: sock.connect((t, int(a))) s = L('on', '%s/%s' % (jid, nick)) except: s = L('off', '%s/%s' % (jid, nick)) msg += '\n%s %s' % (a, s) sock.close() msg = '%s%s' % (t, msg) except: msg = '%s - %s' % (t, L('unknown', '%s/%s' % (jid, nick))) msg = L('Port status at %s', '%s/%s' % (jid, nick)) % msg else: msg = L('What?', '%s/%s' % (jid, nick)) else: msg = L('What?', '%s/%s' % (jid, nick)) send_msg(type, jid, nick, msg) # --------------------------------------------------------------------------- # # REGISTER COMMANDS # # --------------------------------------------------------------------------- # execute = [ (6, 'nslookup', srv_nslookup, 2, 'Command nslookup'), (6, 'host', srv_host, 2, 'Command host'), (6, 'dig', srv_dig, 2, 'Command dig'), (4, 'port', chkserver, 2, 'Check port activity\nport server port1 [port2 ...]'), (4, 'net_ping', net_ping, 2, 'Net Ping.\nnet_ping ip|domain'), (3, 'dns', get_dns, 2, 'DNS resolver.'), (3, 'tld', get_tld, 2, 'Search domain zones TLD.'), ]