From 194815fcaeab6f2195b6ce9942e0f58b1ed722af Mon Sep 17 00:00:00 2001 From: Ponali Date: Sun, 2 Aug 2026 18:29:02 +0200 Subject: [PATCH] basic telnet interface for now it's on par with everything the server supports --- requirements.txt | 3 +- server/main.py | 4 +- telnet-basic/main.py | 619 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 623 insertions(+), 3 deletions(-) create mode 100644 telnet-basic/main.py diff --git a/requirements.txt b/requirements.txt index 5deeb53..9c6802f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,2 @@ -pygments \ No newline at end of file +pygments +prompt_toolkit diff --git a/server/main.py b/server/main.py index 0324a5b..ed8f1f9 100644 --- a/server/main.py +++ b/server/main.py @@ -22,7 +22,7 @@ def makeUserID() -> int: return uid def isValidUsername(name) -> bool: - if len(name)>16: return False + if len(name)<3 or len(name)>16: return False pattern = re.compile("^[a-z][a-z0-9-_]*") match = pattern.match(name) if match==None: return False @@ -87,7 +87,7 @@ async def handleCommand(ws,cmd): sessionCreatedAccount.append(id(ws)) create = time.time() - userdb.insertRow("NAMES",{"id":uid,"name":cmd["name"],"creation_date":create,"last_login":create}) + userdb.insertRow("NAMES",{"id":uid,"name":cmd["name"],"creation_date":create,"last_login":create,"last_namechange":create}) userdb.insertRow("PASSWORDS",{"id":uid,"sum":psum}) return await ws.send(response({"uid":uid})) diff --git a/telnet-basic/main.py b/telnet-basic/main.py new file mode 100644 index 0000000..fee0d27 --- /dev/null +++ b/telnet-basic/main.py @@ -0,0 +1,619 @@ +import sys, os +import asyncio, threading, time, textwrap, re + +import prompt_toolkit as pt +from prompt_toolkit import Application +from prompt_toolkit.buffer import Buffer +from prompt_toolkit.layout.containers import HSplit, VSplit, Window, ConditionalContainer +from prompt_toolkit.layout.controls import BufferControl +from prompt_toolkit.layout.layout import Layout +from prompt_toolkit.layout import FormattedTextControl, Container, is_container, FloatContainer, Float +from prompt_toolkit.widgets import Button, Dialog, Label, TextArea, ValidationToolbar, RadioList +from prompt_toolkit.layout.dimension import Dimension as D +from prompt_toolkit.application.current import get_app +from prompt_toolkit.validation import Validator +from prompt_toolkit.filters import Condition, has_focus, is_true +from prompt_toolkit.key_binding import KeyBindings +from prompt_toolkit.data_structures import Point +from prompt_toolkit.key_binding.bindings.focus import focus_next, focus_previous + +parentdir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.append(parentdir) +from api import ServerConnection + +server = ServerConnection() + +def errorTxt(obj): + if obj["server"]: + return f"Server error: {obj["error"]}" + return f"Client error: {obj["error"]}" + +def isValidUsername(name): + if len(name)<3 or len(name)>16: return False + pattern = re.compile("^[a-z][a-z0-9-_]*") + match = pattern.match(name) + if match==None: return False + return match.start()==0 and match.end()==len(name) + + +def containIfNeeded(el): + if not is_container(el): + return Window(el) + return el + +class TabLayout(): + def __init__(self,kb,tabs,tabFocus,tabNames,name): + self.tabs = tabs + self.tabFocus = tabFocus + self.tabNames = tabNames + + self.current = 0 + self.control = False + self.tabEls = [ConditionalContainer(content=tabs[i],filter=Condition(lambda i=i: self.current==i)) for i in range(len(tabs))] + + # tab bar + self.tbarEls = [Label("StingNET [WIP]"),ConditionalContainer(content=Label("[Ctrl-Up] ",dont_extend_width=True,style="fg:#bbbbbb"),filter=Condition(lambda: not self.control)),ConditionalContainer(content=Label("[Left/Right/Enter] ",dont_extend_width=True),filter=Condition(lambda: self.control))] + self.tbarBtns = [] + for i in range(len(tabNames)): + def handle(i=i): + self.current=i + self.control=False + self.focusTabWindow() + btn = Button(text=tabNames[i],handler=handle,width=16) + self.tbarEls.append(btn) + self.tbarBtns.append(btn) + self.tbarEls=self.tbarEls+[Label(f" {name}",dont_extend_width=True,style="fg:#bbbbbb")] + + self.tbar = VSplit(self.tbarEls,style="bg:#404040") + self.el = HSplit([self.tbar,VSplit(self.tabEls)]) + + # controls + @kb.add("c-up") + def _(e): + self.control = True + self.focusCurrent(e) + + @kb.add("left",filter=Condition(lambda: self.control)) + def _(e): + self.current = (self.current+len(tabNames)-1)%len(tabNames) + self.focusCurrent(e) + + @kb.add("right",filter=Condition(lambda: self.control)) + def _(e): + self.current = (self.current+1)%len(tabNames) + self.focusCurrent(e) + + def focusTabWindow(self): + try: + el = self.tabs[self.current] + if self.tabFocus[self.current]: + el = self.tabFocus[self.current] + get_app().layout.focus(el) + except Exception: + pass + + def focusCurrent(self,e=None): + app = e.app if e else get_app() + app.layout.focus(self.tbarBtns[self.current]) + +class HSelectLayout(): + def __init__(self,kb,opts,optFocus,optNames): + self.opts=opts + self.optFocus=optFocus + self.optNames=optNames + + self.current = 0 + self.select = Window(FormattedTextControl(text=self.show_sel,get_cursor_position=self.curpos,key_bindings=self.make_select_kb()),width=16,dont_extend_height=False,style="bg:#353535") + + self.els = [self.select]+[ConditionalContainer(content=containIfNeeded(opts[i]),filter=Condition(lambda i=i: self.current==i)) for i in range(len(opts))] + self.el = VSplit(self.els) + + self.returnFilter = has_focus(self.el) & ~has_focus(self.select) + @kb.add("c-left",filter=self.returnFilter) + @kb.add("escape",filter=self.returnFilter) + def _(e): + e.app.layout.focus(self.select) + + def make_select_kb(self): + kb = KeyBindings() + + l = len(self.optNames) + + @kb.add("up") + def _(e): + self.current = (self.current+l-1)%l + + @kb.add("down") + def _(e): + self.current = (self.current+1)%l + + @kb.add("enter") + @kb.add("c-right") + @kb.add(" ") + def _(e): + self.focusTabWindow() + + return kb + + def show_sel(self): + o = [] + size = get_app().output.get_size() + height = size.rows + for i in range(len(self.opts)): + name = self.optNames[i] + o.append(("reverse",name) if i==self.current else ("",name)) + o.append((""," \n")) + for i in range(height-len(self.opts)-2): + o.append(("","\n")) + if is_true(self.returnFilter): + o.append(("fg:#bbbbbb","[Ctrl-Left/ESC]")) + return o + + def curpos(self): + return Point(x=len(self.optNames[self.current]),y=self.current) + + def focusTabWindow(self): + if self.optFocus[self.current]: + try: + el = self.optFocus[self.current] + get_app().layout.focus(el) + return + except Exception: + pass + try: + el = self.opts[self.current] + get_app().layout.focus(el) + return + except Exception: + pass + try: + el = self.els[self.current+1] + get_app().layout.focus(el) + return + except Exception: + pass + +quitKB = KeyBindings() + +@quitKB.add("c-q",eager=True) +def exit_(event): + event.app.exit() + +floats = [] + +def spawnDialog(dlg): + global floats + + app = get_app() + + f = Float(content=dlg) + floats.append(f) + prev = app.layout.current_window + app.layout.focus(dlg) + app.invalidate() + + def despawn(): + global floats + floats.remove(f) + app.layout.focus(prev) + app.invalidate() + + return despawn + +async def spawnSimplePopup(title,body): + title = str(title) + if type(body)==str: + body=Label(body) + + res = asyncio.get_running_loop().create_future() + + def handler(): + if not res.done(): + res.set_result(True) + despawn() + + dlg = Dialog( + title=title, + body=body, + buttons=[ + Button("OK",handler=handler) + ] + ) + + despawn = spawnDialog(dlg) + + return await res + +def HButtonSelect(buttonArgs): + kb = KeyBindings() + + # layout = get_app().layout + + btns = [Button(e[0],e[1],width=32) for e in buttonArgs] + + first_selected = has_focus(btns[0]) + last_selected = has_focus(btns[-1]) + + @kb.add("up",filter=~first_selected) + def _(e): + focus_previous(e) + + @kb.add("down",filter=~last_selected) + def _(e): + focus_next(e) + + return HSplit(btns,key_bindings=kb) + + +async def mainScreen(): + kb = KeyBindings() + + @kb.add("c-q",eager=True) + def exit_(event): + event.app.exit() + + stat = await server.status() + if stat["type"]=="error" or not stat["response"]["login"]: return + name = stat["response"]["name"] + + + # ------------------------------------------------------------ + # Settings -> Accounts -> Log Out + # ------------------------------------------------------------ + + def logout(): + get_app().exit(result="login") + + # ------------------------------------------------------------ + # Settings -> Accounts -> Change Name + # ------------------------------------------------------------ + def changename(): + usernameValidator = Validator.from_callable(isValidUsername, error_message='Invalid username') + + def nameaccept(buf: Buffer): + # get_app().layout.focus(okbtn) + dlgclose(True) + return True + + nntxtfield = TextArea( + text=name, + multiline=False, + password=False, + completer=None, + validator=usernameValidator, + accept_handler=nameaccept, + ) + + okbtn = Button("OK",handler=lambda: dlgclose(True)) + cancelbtn = Button("Cancel",handler=lambda: dlgclose(False)) + + async def _changename(): + res = await server.changename(nntxtfield.text) + if res["type"]=="error": + await spawnSimplePopup("Error",errorTxt(res)) + else: + get_app().exit(result="restart") + + def dlgclose(res): + despawn() + if not res: return + get_app().create_background_task(_changename()) + + dlg = Dialog( + title="Change name", + body=HSplit([ + Label(text="Insert your account's new name here:", dont_extend_height=True), + nntxtfield, + Label(text="Keep in mind that you can only change your name every 7 days.", dont_extend_height=True), + ValidationToolbar(), + ]), + buttons=[okbtn, cancelbtn] + ) + + despawn = spawnDialog(dlg) + + # ------------------------------------------------------------ + # Settings -> Accounts -> Change Password + # ------------------------------------------------------------ + def changepass(): + def oldaccept(buf: Buffer): + get_app().layout.focus(newtxtfield) + return True + + def newaccept(buf: Buffer): + # get_app().layout.focus(okbtn) + dlgclose(True) + return True + + oldtxtfield = TextArea( + text="", + multiline=False, + password=True, + completer=None, + validator=None, + accept_handler=oldaccept, + ) + + newtxtfield = TextArea( + text="", + multiline=False, + password=True, + completer=None, + validator=None, + accept_handler=newaccept, + ) + + okbtn = Button("OK",handler=lambda: dlgclose(True)) + cancelbtn = Button("Cancel",handler=lambda: dlgclose(False)) + + async def _changepass(): + res = await server.changepass(oldtxtfield.text,newtxtfield.text) + if res["type"]=="error": + await spawnSimplePopup("Error",errorTxt(res)) + + def dlgclose(res): + despawn() + if not res: return + get_app().create_background_task(_changepass()) + + dlg = Dialog( + title="Change password", + body=HSplit([ + Label(text="Old password:", dont_extend_height=True), + oldtxtfield, + Label(text="New password:", dont_extend_height=True), + newtxtfield, + ValidationToolbar(), + ]), + buttons=[okbtn, cancelbtn] + ) + + despawn = spawnDialog(dlg) + + # ------------------------------------------------------------ + # Settings Menu + # ------------------------------------------------------------ + + settings = HSelectLayout(kb,[ + HButtonSelect([("Log out",logout),("Change username",changename),("Change password",changepass)]) + ],[ + None, + None + ],["Account"]) + + # ------------------------------------------------------------ + # Tab Layout + # ------------------------------------------------------------ + + tablayout = TabLayout(kb,[ + Label("Not implemented yet 1"), + Label("Not implemented yet 2"), + settings.el + ],[ + None, + None, + settings.select + ],["Forum Boards","E-Messages","Settings"],name) + + # ------------------------------------------------------------ + # Application + # ------------------------------------------------------------ + + layout = Layout(FloatContainer(content=tablayout.el,floats=floats),focused_element=tablayout.tabs[0]) + + app = Application(layout=layout, key_bindings=kb, full_screen=True) + app.pre_run_callables.append(tablayout.focusTabWindow) + return await app.run_async() + + +async def createAccountScreen(): + def nameaccept(buf: Buffer) -> bool: + get_app().layout.focus(passtextfield) + return True # Keep text. + + def pwrdaccept(buf: Buffer) -> bool: + # get_app().layout.focus(login_button) + get_app().layout.focus(passtextconffield) + return True # Keep text. + + def pwrdcaccept(buf: Buffer) -> bool: + # get_app().layout.focus(login_button) + ok_handler() + return True # Keep text. + + def _return_none() -> None: + "Button handler that returns None." + get_app().exit() + + def ok_handler() -> None: + get_app().exit(result=(usertextfield.text,passtextfield.text)) + + def login_handler() -> None: + get_app().exit(result="login") + + create_button = Button(text="Create Account", handler=ok_handler, width=16) + login_button = Button(text="Log In", handler=login_handler, width=16) + exit_button = Button(text="Exit", handler=_return_none, width=16) + + usernameValidator = Validator.from_callable(isValidUsername, error_message='Invalid username') + samePassValidator = Validator.from_callable(lambda x: passtextfield.text==x, error_message='Passwords are not the same') + + usertextfield = TextArea( + text="", + multiline=False, + password=False, + completer=None, + validator=usernameValidator, + accept_handler=nameaccept, + ) + + passtextfield = TextArea( + text="", + multiline=False, + password=True, + completer=None, + validator=None, + accept_handler=pwrdaccept, + ) + + passtextconffield = TextArea( + text="", + multiline=False, + password=True, + completer=None, + validator=samePassValidator, + accept_handler=pwrdcaccept, + ) + + dialog = Dialog( + title="Create a StingNET account", + body=HSplit( + [ + Label(text="Username:", dont_extend_height=True), + usertextfield, + Label(text="Password:", dont_extend_height=True), + passtextfield, + Label(text="Password (confirm):", dont_extend_height=True), + passtextconffield, + ValidationToolbar(), + ], + padding=D(preferred=1, max=1), + ), + buttons=[create_button, login_button, exit_button], + with_background=True, + ) + app = Application(key_bindings=quitKB, layout=Layout(dialog), full_screen=True) + + while True: + usertextfield.text = "" + passtextfield.text = "" + passtextconffield.text = "" + + appres = await app.run_async() + if type(appres)==tuple: + name,pwrd = appres + elif type(appres)==str and appres=="login": + return "login" + else: + await server.disconnect() + exit() + + res = await server.createuser(name,pwrd) + if res["type"]=="response": lres = await server.login(name,pwrd) + if res["type"]=="error": + await pt.shortcuts.message_dialog(title="Error (create)",text=f"The account has not been created.\n{errorTxt(res)}").run_async() + if lres["type"]=="error": + await pt.shortcuts.message_dialog(title="Error (login)",text=f"The account has been created, but has not been logged in.\n{errorTxt(lres)}").run_async() + else: + break + +async def loginCredsScreen(): + def nameaccept(buf: Buffer) -> bool: + get_app().layout.focus(passtextfield) + return True # Keep text. + + def pwrdaccept(buf: Buffer) -> bool: + # get_app().layout.focus(login_button) + ok_handler() + return True # Keep text. + + def _return_none() -> None: + "Button handler that returns None." + get_app().exit() + + def ok_handler() -> None: + get_app().exit(result=(usertextfield.text,passtextfield.text)) + + def create_handler() -> None: + get_app().exit(result="create") + + login_button = Button(text="Log In", handler=ok_handler, width=16) + create_button = Button(text="Create Account", handler=create_handler, width=16) + exit_button = Button(text="Exit", handler=_return_none, width=16) + + usernameValidator = Validator.from_callable(isValidUsername, error_message='Invalid username') + + usertextfield = TextArea( + text="", + multiline=False, + password=False, + completer=None, + validator=usernameValidator, + accept_handler=nameaccept, + ) + + passtextfield = TextArea( + text="", + multiline=False, + password=True, + completer=None, + validator=None, + accept_handler=pwrdaccept, + ) + + dialog = Dialog( + title="Log in to StingNET", + body=HSplit( + [ + Label(text="Username:", dont_extend_height=True), + usertextfield, + Label(text="Password:", dont_extend_height=True), + passtextfield, + ValidationToolbar(), + ], + padding=D(preferred=1, max=1), + ), + buttons=[login_button, create_button, exit_button], + with_background=True, + ) + app = Application(key_bindings=quitKB, layout=Layout(dialog), full_screen=True) + + while True: + usertextfield.text = "" + passtextfield.text = "" + + appres = await app.run_async() + if type(appres)==tuple: + name,pwrd = appres + elif type(appres)==str and appres=="create": + return "create" + else: + await server.disconnect() + exit() + + res = await server.login(name,pwrd) + if res["type"]=="error": + await pt.shortcuts.message_dialog(title="Error",text=errorTxt(res)).run_async() + else: + break + +async def logoutScreen(): + while True: + res = await loginCredsScreen() + if res != "create": return + res = await createAccountScreen() + if res != "login": return + +async def main(): + try: + await asyncio.wait_for(server.connect(), timeout=10) + except TimeoutError: + print("The internal server is currently down. Please press Ctrl-C (you may see a KeyboardInterrupt error).") + exit() + + try: + while True: + stat = await server.status() + if stat["type"]=="error": continue + if stat["response"]["login"]: + res = await mainScreen() + if res!="restart": + if res!="login": + break + await server.logout() + else: + await logoutScreen() + finally: + await server.disconnect() + +asyncio.run(main())