diff --git a/.gitignore b/.gitignore index 2ce4b03..bb0f07e 100644 --- a/.gitignore +++ b/.gitignore @@ -222,3 +222,4 @@ __marimo__/ # StingNET server/users.db +server/isettings.db diff --git a/api.py b/api.py index 4480f20..0335545 100644 --- a/api.py +++ b/api.py @@ -155,3 +155,9 @@ class ServerConnection(BasicServerConnection): osum = hashlib.md5(old.encode("utf8")).hexdigest() nsum = hashlib.md5(new.encode("utf8")).hexdigest() return await super().ask({"type": "changepass", "oldpass": osum, "newpass": nsum}) + + async def getisettings(self,interface: str): + return await super().ask({"type": "getisettings","interface": interface}) + + async def setisettings(self,interface: str,settings: dict): + return await super().ask({"type": "setisettings","interface": interface,"settings":settings}) diff --git a/server/main.py b/server/main.py index ed8f1f9..ad3583d 100644 --- a/server/main.py +++ b/server/main.py @@ -12,6 +12,9 @@ userdb = Database("users") userdb.createTable("NAMES",{"id":"int","name":"string","creation_date":"int","last_login":"int","last_namechange":"int"}) userdb.createTable("PASSWORDS",{"id":"int","sum":"string"}) +isetdb = Database("isettings") +isetdb.createTable("BASIC",{"id":"int","color":"int","encoding":"str"}) + def makeUserID() -> int: count = userdb.rowAmount("NAMES") digits = max(4,1+int(math.log10(10+count))) @@ -35,6 +38,13 @@ def isValidChecksum(cs) -> bool: if match==None: return False return match.start()==0 and match.end()==len(cs) +def isValidSimpleName(name) -> bool: + if len(name)==0: return False + pattern = re.compile("^[a-z][a-z0-9-_]*") + match = pattern.match(name.lower()) + if match==None: return False + return match.start()==0 and match.end()==len(name) + def response(msg,**kwargs) -> str: return json.dumps({"type":"response","response":msg,**kwargs}) @@ -154,6 +164,37 @@ async def handleCommand(ws,cmd): userdb.updateFromPK("PASSWORDS","id",uid,{"sum":newpass}) return await ws.send(response("success")) + # interface settings + case "getisettings": + if uid == None: + return await ws.send(clientError("You are not logged in.")) + intname = cmd["interface"] + intname = intname.upper() + if (not isValidSimpleName(intname)) or (not isetdb.tableExists(intname)): + return await ws.send(clientError(f"The interface '{intname}' is not an existing interface.")) + row = isetdb.rowFromPK(intname,"id",uid) + if row == None: + return await ws.send(clientError("This user has no settings for this interface set up.")) + del row["id"] + return await ws.send(response(row)) + + case "setisettings": + if uid == None: + return await ws.send(clientError("You are not logged in.")) + intname = cmd["interface"] + intname = intname.upper() + if (not isValidSimpleName(intname)) or (not isetdb.tableExists(intname)): + return await ws.send(clientError(f"The interface '{intname}' is not an existing interface.")) + newset = cmd["settings"] + if not isetdb.hasPK(intname,"id",uid): + newset["id"]=uid + isetdb.insertRow(intname,newset) + else: + if "id" in newset: + del newset["id"] + isetdb.updateFromPK(intname,"id",uid,newset) + return await ws.send(response("success")) + case _: return await ws.send(clientError(f"Command '{cmd["type"]}' does not exist.")) diff --git a/telnet-basic/main.py b/telnet-basic/main.py index fee0d27..20121bb 100644 --- a/telnet-basic/main.py +++ b/telnet-basic/main.py @@ -1,5 +1,6 @@ -import sys, os +import sys, os, io import asyncio, threading, time, textwrap, re +import encodings.aliases import prompt_toolkit as pt from prompt_toolkit import Application @@ -7,15 +8,17 @@ 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 import FormattedTextControl, Container, is_container, FloatContainer, Float, WindowAlign +from prompt_toolkit.widgets import Box, 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.key_binding import KeyBindings, merge_key_bindings +from prompt_toolkit.key_binding.defaults import load_key_bindings from prompt_toolkit.data_structures import Point from prompt_toolkit.key_binding.bindings.focus import focus_next, focus_previous +from prompt_toolkit.output import ColorDepth, create_output parentdir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.append(parentdir) @@ -36,11 +39,96 @@ def isValidUsername(name): return match.start()==0 and match.end()==len(name) +intset = {} # interface settings + +def getColorDepth(): + try: + return ColorDepth(f"DEPTH_{intset["color"]}_BIT") + except ValueError: + return ColorDepth.DEPTH_4_BIT + +def defaultAppSettings(): + return {"full_screen":True, "color_depth":getColorDepth(), "output": createEncodedOutput(intset["encoding"])} + +class EncodingProxy(io.TextIOBase): + def __init__(self, raw, encoding: str): + self._raw = raw + self._encoding = encoding + # self.buffer = raw.buffer + self.pipeReplace = False + for ch in ["┌","┐","┘","└","─","│"]: + if ch.encode(encoding,errors="replace").decode(encoding)!=ch: + self.pipeReplace = True + break + + @property + def encoding(self) -> str: + return self._encoding + + def write(self, s: str) -> int: + if self.encoding=="ascii" or self.pipeReplace: + for corner in ["┌","┐","┘","└"]: + s=s.replace(corner,"+") + s=s.replace("─","-") + s=s.replace("│","|") + pass + + byts = s.encode(self._encoding,errors="replace") + # print(s.encode("utf8"),byts) + self._raw.buffer.write(bytes(byts)) + self._raw.flush() + return len(s) + + def flush(self) -> None: + self._raw.flush() + + def fileno(self): + return self._raw.fileno() + + def isatty(self): + return True + +def createEncodedOutput(enc): + proxy: TextIO = EncodingProxy(sys.stdout, encoding=enc) + return create_output(proxy) + def containIfNeeded(el): if not is_container(el): return Window(el) return el +def message_dialog( # derived from prompt_toolkit's code + title = "", + text = "", + ok_text: str = "Ok", + style = None, +) -> Application[None]: + """ + Display a simple message box and wait until the user presses enter. + """ + def _return_none() -> None: + "Button handler that returns None." + get_app().exit() + + dialog = Dialog( + title=title, + body=Label(text=text, dont_extend_height=True), + buttons=[Button(text=ok_text, handler=_return_none)], + with_background=True, + ) + + # Key bindings. + bindings = KeyBindings() + bindings.add("tab")(focus_next) + bindings.add("s-tab")(focus_previous) + + return Application( + layout=Layout(dialog), + key_bindings=merge_key_bindings([load_key_bindings(), bindings]), + style=style, + **defaultAppSettings() + ) + class TabLayout(): def __init__(self,kb,tabs,tabFocus,tabNames,name): self.tabs = tabs @@ -52,7 +140,7 @@ class TabLayout(): 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.tbarEls = [ConditionalContainer(Label("StingNET [WIP]"),Condition(lambda: get_app().output.get_size().columns >= 90)),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): @@ -69,6 +157,7 @@ class TabLayout(): # controls @kb.add("c-up") + @kb.add("c-j") def _(e): self.control = True self.focusCurrent(e) @@ -110,6 +199,7 @@ class HSelectLayout(): self.returnFilter = has_focus(self.el) & ~has_focus(self.select) @kb.add("c-left",filter=self.returnFilter) + @kb.add("c-h",filter=self.returnFilter) @kb.add("escape",filter=self.returnFilter) def _(e): e.app.layout.focus(self.select) @@ -129,6 +219,7 @@ class HSelectLayout(): @kb.add("enter") @kb.add("c-right") + @kb.add("c-l") @kb.add(" ") def _(e): self.focusTabWindow() @@ -229,7 +320,7 @@ def HButtonSelect(buttonArgs): # layout = get_app().layout - btns = [Button(e[0],e[1],width=32) for e in buttonArgs] + btns = [Button(e[0],e[1],width=48) for e in buttonArgs] first_selected = has_focus(btns[0]) last_selected = has_focus(btns[-1]) @@ -370,16 +461,149 @@ async def mainScreen(): despawn = spawnDialog(dlg) + # ------------------------------------------------------------ + # Settings -> Interface -> Change color depth + # ------------------------------------------------------------ + + async def _updateisettings(): + res = await server.setisettings("basic",intset) + if res["type"]=="error": + if res["server"]: + await spawnSimplePopup("Error",f"{errorTxt(res)}\nYour changes might not have been sent to the internal server. Please try again.") + else: + await spawnSimplePopup("Error",errorTxt(res)) + + def changecolor(): + def dlgclose(): + despawn() + intset["color"] = colorselect.current_value + get_app().create_background_task(_updateisettings()) + + ok_button = Button(text="OK", handler=dlgclose, width=16) + + colorselect = RadioList([(24,"24 bits (True Color)"),(8,"8 bits (Paletted)"),(4,"4 bits (ANSI/RGBI)"),(1,"1 bit (Monochrome)")],default=intset["color"]) + + dlg = Dialog( + title="Change color depth", + body=HSplit( + [ + Label(text="Please select a color depth supported by this terminal:", dont_extend_height=True), + colorselect, + Label(text="Use 1 or 4 bits for compatibility with older terminals, otherwise 8 or 24 bits.", dont_extend_height=True), + Label(text="The more bits in the color depth, the more colors there can be displayed on-screen."), + ValidationToolbar(), + ], + padding=D(preferred=1, max=1), + ), + buttons=[ok_button], + with_background=True, + ) + + despawn = spawnDialog(dlg) + + def changeencother(): + def dlgclose(): + despawn() + intset["encoding"] = encselect.current_value + get_app().create_background_task(_updateisettings()) + + ok_button = Button(text="OK", handler=dlgclose, width=16) + + encodingList = sorted(set(encodings.aliases.aliases.values())) + for e in encodingList.copy(): + try: + ("test").encode(e,errors="replace") + except LookupError: + encodingList.remove(e) + encselect = RadioList([(e,e) for e in encodingList],default=intset["encoding"]) + + dlg = Dialog( + title="Change text encoding", + body=HSplit( + [ + Label(text="Please select the encoding supported by this terminal:", dont_extend_height=True), + encselect, + Label(text="For more information about every supported encoding, open this webpage:\nhttps://docs.python.org/3.14/library/codecs.html#standard-encodings",dont_extend_height=True), + ValidationToolbar(), + ], + padding=D(preferred=1, max=1), + ), + buttons=[ok_button], + with_background=True, + ) + + despawn = spawnDialog(dlg) + + def changeenc(): + def dlgclose(): + despawn() + v = encselect.current_value + if v=="other": + changeencother() + else: + intset["encoding"] = v + get_app().create_background_task(_updateisettings()) + + ok_button = Button(text="OK", handler=dlgclose, width=16) + + default = intset["encoding"] + if not default in ["utf8","ascii","latin_1","cp437"]: + default = "other" + + encselect = RadioList([("utf8","UTF-8 (Unicode)"),("ascii","ASCII (7-bit)"),("latin_1","Latin-1/ISO-8859-1 (Ext. ASCII)"),("cp437","CP437 (Extended ASCII)"),("other","Other (Python-supported)")],default=default) + + dlg = Dialog( + title="Change text encoding", + body=HSplit( + [ + Label(text="Please select the encoding supported by this terminal:", dont_extend_height=True), + encselect, + Label(text="UTF-8 is generally reccomended for a normal modern terminal, and supports everything from Unicode.",dont_extend_height=True), + Label(text="ASCII is very barebones but is supported by every terminal. Please use it if all else fails.", dont_extend_height=True), + Label(text="You may see garbled text if you chose the wrong text encoding.",dont_extend_height=True), + ValidationToolbar(), + ], + padding=D(preferred=1, max=1), + ), + buttons=[ok_button], + with_background=True, + ) + + despawn = spawnDialog(dlg) + + # ------------------------------------------------------------ + # Settings -> Interface (Status) + # ------------------------------------------------------------ + + def showISettings(): + color = ({1: "1 bit (Monochrome)",4: "4 bits (ANSI/RGBI)",8: "8 bits",24: "24 bits (True Color)"})[intset["color"]] + encoding = intset["encoding"] + if encoding=="utf8": encoding="UTF-8 (Unicode)" + if encoding=="ascii": encoding="ASCII (7-bit)" + if encoding=="cp437": encoding="CP437 (Extended ASCII)" + if encoding=="latin_1": encoding="Latin-1/ISO-8859-1 (Ext. ASCII)" + + return [ + ("","Color depth: "),("bold",color),("","\n"), + ("","Text encoding: "),("bold",encoding),("","\n") + ] + # ------------------------------------------------------------ # Settings Menu # ------------------------------------------------------------ settings = HSelectLayout(kb,[ - HButtonSelect([("Log out",logout),("Change username",changename),("Change password",changepass)]) + HButtonSelect([("Log out",logout),("Change username",changename),("Change password",changepass)]), + + HSplit([ + Window(FormattedTextControl(text=showISettings,focusable=False),dont_extend_height=True), + Box(HButtonSelect([("Change color depth",changecolor),("Change text encoding",changeenc)]),padding=0), + Label("\n\nYou will need to restart or relog to apply any new changes."), + ]) ],[ None, None - ],["Account"]) + ],["Account","Interface"]) # ------------------------------------------------------------ # Tab Layout @@ -401,11 +625,59 @@ async def mainScreen(): layout = Layout(FloatContainer(content=tablayout.el,floats=floats),focused_element=tablayout.tabs[0]) - app = Application(layout=layout, key_bindings=kb, full_screen=True) + app = Application(layout=layout, key_bindings=kb, **defaultAppSettings()) app.pre_run_callables.append(tablayout.focusTabWindow) return await app.run_async() +async def firstSetupScreen(): + def ok_handler() -> None: + get_app().exit() + + ok_button = Button(text="OK", handler=ok_handler, width=16) + + colorselect = RadioList([(24,"24 bits (True Color)"),(8,"8 bits (Paletted)"),(4,"4 bits (ANSI/RGBI)"),(1,"1 bit (Monochrome)")],default=8) + + encselect = RadioList([("utf8","UTF-8 (Unicode)"),("ascii","ASCII (7-bit)"),("latin_1","ISO-8859-1 (Ext. ASCII)"),("cp437","CP437 (Extended ASCII)")],default="utf8") + + cdialog = Dialog( + title="First Setup", + body=HSplit( + [ + Label(text="Please select a color depth supported by this terminal:", dont_extend_height=True), + colorselect, + Label(text="Use 1 or 4 bits for compatibility with older terminals, otherwise 8 or 24 bits.", dont_extend_height=True), + Label(text="You can always change this later in the interface settings.",dont_extend_height=True), + ValidationToolbar(), + ], + padding=D(preferred=1, max=1), + ), + buttons=[ok_button], + with_background=True, + ) + capp = Application(layout=Layout(cdialog), **defaultAppSettings()) + await capp.run_async() + + edialog = Dialog( + title="First Setup", + body=HSplit( + [ + Label(text="Please insert the text encoding supported by this terminal:", dont_extend_height=True), + encselect, + Label(text="Use ASCII for compatibility with older terminals, otherwise UTF-8.", dont_extend_height=True), + Label(text="You can always change this later in the interface settings.",dont_extend_height=True), + ValidationToolbar(), + ], + padding=D(preferred=1, max=1), + ), + buttons=[ok_button], + with_background=True, + ) + eapp = Application(layout=Layout(edialog), **defaultAppSettings()) + await eapp.run_async() + + return {"color":colorselect.current_value,"encoding":encselect.current_value} + async def createAccountScreen(): def nameaccept(buf: Buffer) -> bool: get_app().layout.focus(passtextfield) @@ -482,7 +754,7 @@ async def createAccountScreen(): buttons=[create_button, login_button, exit_button], with_background=True, ) - app = Application(key_bindings=quitKB, layout=Layout(dialog), full_screen=True) + app = Application(key_bindings=quitKB, layout=Layout(dialog), **defaultAppSettings()) while True: usertextfield.text = "" @@ -501,9 +773,9 @@ async def createAccountScreen(): 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() + await 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() + await 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 @@ -566,7 +838,7 @@ async def loginCredsScreen(): buttons=[login_button, create_button, exit_button], with_background=True, ) - app = Application(key_bindings=quitKB, layout=Layout(dialog), full_screen=True) + app = Application(key_bindings=quitKB, layout=Layout(dialog), **defaultAppSettings()) while True: usertextfield.text = "" @@ -583,7 +855,7 @@ async def loginCredsScreen(): res = await server.login(name,pwrd) if res["type"]=="error": - await pt.shortcuts.message_dialog(title="Error",text=errorTxt(res)).run_async() + await message_dialog(title="Error",text=errorTxt(res)).run_async() else: break @@ -595,6 +867,8 @@ async def logoutScreen(): if res != "login": return async def main(): + global intset + try: await asyncio.wait_for(server.connect(), timeout=10) except TimeoutError: @@ -606,12 +880,21 @@ async def main(): stat = await server.status() if stat["type"]=="error": continue if stat["response"]["login"]: + ires = await server.getisettings("basic") + if ires["type"]=="error": + intset = await firstSetupScreen() + ires = await server.setisettings("basic",intset) + if ires["type"]=="error": + await message_dialog(title="Error",text=f"{errorTxt(ires)}\n\nThese settings might not have been sent to the internal server.\nYou will be asked again the next time you log in to your account.").run_async() + else: + intset = ires["response"] res = await mainScreen() if res!="restart": if res!="login": break await server.logout() else: + intset = {"color":4,"encoding":"ascii"} await logoutScreen() finally: await server.disconnect()