Recoded Base

Author: SudoLite

common/__init__.py

@@ -1,4 +1,8 @@
 import asyncio
+from asyncio import sleep, Event, create_task
+from contextlib import suppress
+from getpass import getpass
+from itertools import cycle
 from os import name, system
 from socket import gethostname
 
@@ -9,10 +13,15 @@
 from tools.impl import handle as tools_handle
 
 
+# noinspection PyMethodMayBeStatic
 class Console:
-    @staticmethod
-    async def banner():
-        await Console.clear()
+    def __init__(self):
+        self.using = None
+        self.dots = cycle(["|", "/", "-", "\\"])
+        self.loading_text = ""
+
+    async def banner(self):
+        await self.clear()
         await aprint(Colorate.Diagonal(Colors.yellow_to_red, Center.XCenter("""
 ╔════════════════════════════════════════════════════════════════════════════════╗
 ║    ooooooooo.              ooooooooooooo                     oooo              ║
@@ -28,37 +37,54 @@ async def banner():
                         ║   Coded By : MXTeam - V 1.0     ║
                         ╚═════════════════════════════════╝\n\n\n\n""")))
 
-    @staticmethod
-    async def input(*messages):
-        return await ainput(' '.join([txt for txt in messages]))
+    async def loader(self):
+        while 1:
+            if not self.using.is_set():
+                await self.using.wait()
+            await aprint(Colorate.Horizontal(Colors.rainbow,
+                                             "[%s] %s" %
+                                             (next(self.dots),
+                                              self.loading_text)),
+                         end="\r")
+            await sleep(.05)
+
+    async def input(self, *messages, hide_value=False):
+        if not hide_value:
+            return await ainput(' '.join([str(txt) for txt in messages]))
+        return getpass(' '.join([str(txt) for txt in messages]))
 
-    @staticmethod
-    async def error(*messages):
-        await aprint(Colorate.Diagonal(Colors.red_to_white, "[X] " + ' '.join([txt for txt in messages])))
+    async def cinput(self, *messages, hide_value=False):
+        return await self.input(Colorate.Horizontal(Colors.yellow_to_red,
+                                                    f"╔═══[{gethostname()}@PyTools]"
+                                                    f"{(' (' + ' '.join([txt for txt in messages]) + ')') if messages else ''} "
+                                                    f"\n╚══════> "), hide_value=hide_value)
 
-    @staticmethod
-    async def clear():
-        await Console.command('cls' if name == 'nt' else 'clear')
+    async def error(self, *messages):
+        await aprint(Colorate.Diagonal(Colors.red_to_white, "[X] " + ' '.join([str(txt) for txt in messages])))
 
-    @staticmethod
-    async def run():
+    async def clear(self):
+        if self.using:
+            self.using.clear()
+            await aprint(' ' * len(self.loading_text), end="\r")
+            self.loading_text = ''
+        await self.command('cls' if name == 'nt' else 'clear')
+
+    async def run(self):
+        self.using = Event()
+        create_task(self.loader())
         while 1:
-            inp = (await Console.input(Colorate.Horizontal(Colors.yellow_to_red,
-                                                           f"╔═══[{gethostname()}"
-                                                           f"@PyTools]\n╚══════> ")))
+            inp = await self.cinput()
             if not inp:
                 pass
-            await Console.handle(inp.strip())
+            await self.handle(inp.strip())
+            self.using.clear()
 
-    @staticmethod
-    async def command(cmd):
-        try:
+    async def command(self, cmd):
+        with suppress(Exception):
             return await (await asyncio.create_subprocess_shell(cmd=cmd)).communicate()
-        except:
-            system(cmd)
+        system(cmd)
 
-    @staticmethod
-    async def handle(inp):
+    async def handle(self, inp):
         cmd, args = tools.Patterns.parsecommand(inp)
         cmd = cmd.upper()
 
@@ -68,16 +94,15 @@ async def handle(inp):
             elif {cmd} & {"EXIT", "QUIT", "LOGOUT", "EXIT()"}:
                 exit(-1)
             elif {cmd} & {"CLEAR", "CLS"}:
-                await Console.banner()
-            elif await tools_handle(cmd, *args):
+                await self.banner()
+            elif await tools_handle(self, cmd, *args):
                 pass
             elif cmd:
-                await Console.error("\"%s\" command not found" % cmd.lower())
-
+                await self.error("\"%s\" command not found" % cmd.lower())
 
         except Exception as e:
-            await Console.error(str(e) or repr(e))
+            self.using.clear()
+            await self.error((str(e) or repr(e)) + ' ' * 50)
 
-    @staticmethod
-    async def info(*messages):
+    async def info(self, *messages):
         await aprint(Colorate.Diagonal(Colors.blue_to_white, "[!] " + ' '.join([txt for txt in messages])))

main.py

@@ -4,8 +4,9 @@
 from common import Console
 
 if __name__ == '__main__':
-    run(Console.banner())
+    console = Console()
+    run(console.banner())
     while 1:
         with suppress(KeyboardInterrupt):
             print()
-            run(Console.run())
+            run(console.run())

tools/__init__.py

@@ -9,7 +9,7 @@
 from typing import Callable, Any, List
 from time import time
 
-__all__ = ["Patterns", "Random", "Tool", "Math"]
+__all__ = ["Patterns", "Random", "Tool", "Math", "Timer"]
 
 
 class Random:
@@ -67,7 +67,8 @@ def parsecommand(cmd: str):
 
 
 class Tool:
-    async def run(self, *args):
+    @staticmethod
+    async def run(console, *args):
         raise NotImplementedError
 
 
@@ -78,6 +79,7 @@ class Math:
     def ping_sizer(lists):
         return (min(lists, key=int), max(lists, key=int), round(sum(lists) / len(lists), 2)) if lists else (0, 0, 0)
 
+
 class Timer:
     _start: time
     _done: time

tools/impl/Pinger.py

@@ -1,6 +1,5 @@
-from asyncio import sleep, open_connection, TimeoutError, wait_for, create_task
+from asyncio import sleep, open_connection, TimeoutError, wait_for
 from contextlib import suppress
-from itertools import cycle
 
 from aioconsole import aprint
 from aiohttp import ClientSession
@@ -10,15 +9,11 @@
 
 from tools import Tool, Timer
 
-dots = cycle(["|", "/", "-", "\\"])
-
 
 # noinspection PyUnusedLocal
 class Pinger(Tool):
-    using = False
-
     @staticmethod
-    async def run(*args):
+    async def run(console, *args):
         assert len(args) == 2, "bad args"
 
         bad = 0
@@ -33,33 +28,33 @@ async def run(*args):
         assert method, "invalid method"
 
         address = target.human_repr() if method == Pinger.HTTP else target.host
-        try:
-            with suppress(KeyboardInterrupt):
-                Pinger.using = True
-                create_task(Pinger.amim(args[0].upper(), address, port))
-
-                while 1:
-                    request = await method(address, port)
-                    if not request[1] and request[1] != 408:
-                        bad += 1
-
-                    counter += 1
-
-                    await sleep(0.5)
-                    pings.append(request[0])
-
-                    await aprint(
-                        Colorate.Horizontal(
-                            Colors.green_to_cyan if request[1] and request[1] != 408 else Colors.red_to_purple,
-                            "[%s] Reply from %s%sstatus %s protocol %s time: %sms" % (
-                                counter,
-                                address,
-                                (f" port {port} " if method not in [Pinger.ICMP, Pinger.HTTP] else " "),
-                                Pinger.status(request, method),
-                                args[0].upper(),
-                                request[0])))
-        finally:
-            Pinger.using = False
+        with suppress(KeyboardInterrupt):
+            console.using.set()
+            console.loading_text = "Pinging %s%susing %s protocol" % (address,
+                                                                      (f" port {port} " if args[0].upper() not in [
+                                                                          "ICMP", "HTTP"] else " "),
+                                                                      args[0].upper())
+
+            while 1:
+                request = await method(address, port)
+                if not request[1] and request[1] != 408:
+                    bad += 1
+
+                counter += 1
+
+                await sleep(0.5)
+                pings.append(request[0])
+
+                await aprint(
+                    Colorate.Horizontal(
+                        Colors.green_to_cyan if request[1] and request[1] != 408 else Colors.red_to_purple,
+                        "[%s] Reply from %s%sstatus %s protocol %s time: %sms" % (
+                            counter,
+                            address,
+                            (f" port {port} " if method not in [Pinger.ICMP, Pinger.HTTP] else " "),
+                            Pinger.status(request, method),
+                            args[0].upper(),
+                            request[0])))
 
     @staticmethod
     async def ICMP(ip, port):
@@ -116,15 +111,3 @@ def status(request, method):
         else:
             s = "OVERLOAD"
         return s
-
-    @staticmethod
-    async def amim(method, target, port):
-        while Pinger.using:
-            await aprint(
-                "[%s] Pinging %s%susing %s protocol" %
-                (next(dots),
-                 target,
-                 (f" port {port} " if method not in ["ICMP", "HTTP"] else " "),
-                 method),
-                end="\r")
-            await sleep(.05)

tools/impl/SSH.py

@@ -1,32 +1,24 @@
 from contextlib import suppress
-from getpass import getpass
 from aioconsole import aprint
 from asyncssh import SSHClientConnection, connect
 from tools import Tool
 
 
 class SSH(Tool):
-    using = False
-
     @staticmethod
-    async def run(*args):
+    async def run(console, *args):
         assert len(args) == 2, "bad args"
-        ip, username = str(args[0]).split(":"), args[1]  # SSH 185.855.855.690:22 root
-        password = getpass("Enter Password: ")
-        await aprint(f"Connecting To {ip[0]}")
-        try:
-            with suppress(KeyboardInterrupt):
-                SSH.using = True
-                try:
-                    Connection: SSHClientConnection
+        ip, username = str(args[0]).split(":"), args[1]
+        password = await console.cinput("Enter Password", hide_value=True)
+        console.using.set()
+        console.loading_text = f"Connecting To {ip[0]}"
+        with suppress(KeyboardInterrupt):
+            Connection: SSHClientConnection
 
-                    async with connect(ip[0], port=int(ip[1]), username=username, password=password,
-                                       known_hosts=None, connect_timeout=5) as Connection:
-                        while True:
-                            inputcmd = input(f"{username}@{ip[0]}> ")
-                            output = await Connection.run(inputcmd)
-                            await aprint(output.stdout)
-                except Exception as e:
-                    await aprint(str(e))
-        finally:
-            SSH.using = False
+            async with connect(ip[0], port=int(ip[1]), username=username, password=password,
+                               known_hosts=None, connect_timeout=5) as Connection:
+                await console.banner()
+                while True:
+                    inputcmd = await console.cinput(f"{username}@{ip[0]}")
+                    output = await Connection.run(inputcmd)
+                    await aprint(output.stdout)

tools/impl/__init__.py

@@ -5,14 +5,14 @@
 __all__ = ["handle"]
 
 
-async def handle(cmd, *args):
+async def handle(console, cmd, *args):
     if {cmd} & {"PING", "PINGER"}:
-        await Pinger.run(*args)
+        await Pinger.run(console, *args)
         return True
     elif {cmd} & {"CFX"}:
-        await Cfxfinder.run(*args)
+        await Cfxfinder.run(console, *args)
         return True
     elif {cmd} & {"SSH"}:
-        await SSH.run(*args)
+        await SSH.run(console, *args)
         return True
     return False