From a83ad194ebb0f0f2a41322806792d63d5645d96a Mon Sep 17 00:00:00 2001 From: Ponali Date: Fri, 7 Aug 2026 16:44:28 +0200 Subject: [PATCH] added admins and banning this took more time than i thought it would --- api.py | 26 ++++-- requirements.txt | 1 + server/main.py | 92 ++++++++++++++++++++- telnet-basic/main.py | 189 ++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 299 insertions(+), 9 deletions(-) diff --git a/api.py b/api.py index 0335545..82837bf 100644 --- a/api.py +++ b/api.py @@ -125,19 +125,23 @@ class ServerConnection(BasicServerConnection): async def status(self): return await super().ask({"type": "status"}) - async def __login(self): + async def __login(self,logincmd=None): # print("__login") - if self.logincmd == None: return {"type": "error", "error": "No credentials found.", "server": False} + if logincmd==None: logincmd=self.logincmd + if logincmd == None: return {"type": "error", "error": "No credentials found.", "server": False} stat = await self.status() if stat==None: return if stat["type"]=="error" or stat["response"]["login"]==False: - return await super().ask(self.logincmd) + return await super().ask(logincmd) async def login(self,name: str,pwrd: str): res = hashlib.md5(pwrd.encode("utf8")) psum = res.hexdigest() - self.logincmd = {"type": "login", "name": name, "pass": psum} - return await self.__login() + logincmd = {"type": "login", "name": name, "pass": psum} + res = await self.__login(logincmd) + if res["type"]!="error": + self.logincmd = logincmd + return res async def logout(self): self.logincmd = None @@ -156,8 +160,20 @@ class ServerConnection(BasicServerConnection): nsum = hashlib.md5(new.encode("utf8")).hexdigest() return await super().ask({"type": "changepass", "oldpass": osum, "newpass": nsum}) + async def getuserinfo(self,name_or_id: str|int): + if type(name_or_id)==str: + return await super().ask({"type": "getuserinfo","name": name_or_id}) + else: + return await super().ask({"type": "getuserinfo","id": name_or_id}) + 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}) + + async def banuser(self,uid: str,duration: int=0,motive: None|str=None): + return await super().ask({"type": "banuser","user": uid,"duration": duration,"motive": motive}) + + async def unbanuser(self,uid: str): + return await super().ask({"type": "unbanuser","user": uid}) diff --git a/requirements.txt b/requirements.txt index 9c6802f..e387842 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ +humanize pygments prompt_toolkit diff --git a/server/main.py b/server/main.py index ad3583d..23bae09 100644 --- a/server/main.py +++ b/server/main.py @@ -1,7 +1,7 @@ #!/usr/bin/env python import asyncio -import json, random, re, math, time, traceback +import json, random, re, math, time, traceback, datetime, humanize from websockets.exceptions import ConnectionClosed, ConnectionClosedOK from websockets.asyncio.server import serve @@ -11,6 +11,8 @@ from db import Database 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"}) +userdb.createTable("ADMIN",{"id":"int","is_admin":"int"}) +userdb.createTable("BANNED",{"id":"int","is_banned":"int","motive":"str","ban_date":"int","expire_date":"int"}) isetdb = Database("isettings") isetdb.createTable("BASIC",{"id":"int","color":"int","encoding":"str"}) @@ -65,6 +67,28 @@ def forgetSession(sid): if sid in sessionCreatedAccount: sessionCreatedAccount.remove(sid) +def isAdmin(uid): + try: + if uid == None: return False + if not userdb.tableExists("ADMIN"): return False + row = userdb.rowFromPK("ADMIN","id",uid) + if row==None: return False + if int(row["is_admin"])!=1: return False + return True + except Exception: + return False + +def isBanned(uid): + if not userdb.tableExists("BANNED"): return False + row = userdb.rowFromPK("BANNED","id",uid) + if row==None: return False + if int(row["is_banned"])==1: + if row["expire_date"]!=0 and time.time()>row["expire_date"]: + userdb.updateFromPK("BANNED","id",uid,{"is_banned":0}) + return False + return True + return False + async def handleCommand(ws,cmd): global sessionUsers print(f"recieved command: {cmd} (from {id(ws)})") @@ -117,6 +141,11 @@ async def handleCommand(ws,cmd): dbsum = userdb.rowFromPK("PASSWORDS","id",uid)["sum"] if psum.lower() != dbsum.lower(): return await ws.send(clientError("Invalid username or password.")) + if isBanned(uid): + row = userdb.rowFromPK("BANNED","id",uid) + expire = row["expire_date"] + timemsg = "You are permanently banned." if expire==0 else f"You will be unbanned in {humanize.naturaltime(datetime.datetime.fromtimestamp(expire))}." + return await ws.send(clientError(f"Your account has been banned!\nMovite: {row["motive"]}\n\n{timemsg}")) print(f"session {id(ws)} has logged in as {name} ({uid})") sessionUsers[id(ws)] = uid userdb.updateFromPK("NAMES","id",uid,{"last_login":time.time()}) @@ -132,7 +161,7 @@ async def handleCommand(ws,cmd): if uid == None: return await ws.send(response({"login":False})) name = userdb.rowFromPK("NAMES","id",uid)["name"] - return await ws.send(response({"login":True,"uid":uid,"name":name})) + return await ws.send(response({"login":True,"uid":uid,"name":name,"admin":isAdmin(uid)})) case "changename": if uid == None: @@ -164,6 +193,35 @@ async def handleCommand(ws,cmd): userdb.updateFromPK("PASSWORDS","id",uid,{"sum":newpass}) return await ws.send(response("success")) + case "getuserinfo": + if "name" in cmd: + name = cmd["name"] + row = userdb.findRow("NAMES","name = ?",(name,)) + if not row: await ws.send(clientError("Cannot find this user.")) + auid = row["id"] + else: + auid = row["id"] + row = userdb.rowFromPK("NAMES","id",auid) + if not row: await ws.send(clientError("Cannot find this user.")) + name = row["name"] + banned = isBanned(auid) + ban_details = {} + if banned: + brow = userdb.rowFromPK("BANNED","id",auid) + ban_details["motive"] = brow["motive"] + ban_details["ban_date"] = brow["ban_date"] + ban_details["expire_date"] = brow["expire_date"] + + return await ws.send(response({ + "name": name, + "id": auid, + "created": row["creation_date"], + "last_login": row["last_login"], + "last_namechange": row["last_namechange"], + "banned": banned, + "ban_details": ban_details + })) + # interface settings case "getisettings": if uid == None: @@ -195,6 +253,36 @@ async def handleCommand(ws,cmd): isetdb.updateFromPK(intname,"id",uid,newset) return await ws.send(response("success")) + # admin-specific + case "banuser": + if not isAdmin(uid): + return await ws.send(clientError("This account does not have admin privileges.")) + user = cmd["user"] + if not userdb.hasPK("NAMES","id",user): + return await ws.send(clientError("Cannot find this user.")) + motive = cmd["motive"] + duration = cmd["duration"] # in seconds + expire = time.time()+duration if duration>0 else 0 + if userdb.hasPK("BANNED","id",user): + userdb.updateFromPK("BANNED","id",user,{"is_banned":1,"motive":motive,"ban_date": time.time(),"expire_date":expire}) + else: + userdb.insertRow("BANNED",{"id":user,"is_banned":1,"motive":motive,"ban_date": time.time(),"expire_date":expire}) + return await ws.send(response("success")) + + case "unbanuser": + if not isAdmin(uid): + return await ws.send(clientError("This account does not have admin privileges.")) + user = cmd["user"] + if not userdb.hasPK("NAMES","id",user): + return await ws.send(clientError("Cannot find this user.")) + if userdb.hasPK("BANNED","id",user): + if userdb.rowFromPK("BANNED","id",user)["is_banned"]==0: + return await ws.send(clientError("This user is not banned.")) + userdb.updateFromPK("BANNED","id",user,{"is_banned":0}) + else: + userdb.insertRow("BANNED",{"id":user,"is_banned":0,"motive":""}) + 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 74a5134..33a0bf9 100644 --- a/telnet-basic/main.py +++ b/telnet-basic/main.py @@ -1,5 +1,6 @@ import sys, os, io import asyncio, threading, time, textwrap, re +import humanize, datetime import encodings.aliases import prompt_toolkit as pt @@ -317,6 +318,31 @@ async def spawnSimplePopup(title,body): return await res +async def spawnYesNoPopup(title,body): + title = str(title) + if type(body)==str: + body=Label(body) + + res = asyncio.get_running_loop().create_future() + + def handler(r): + if not res.done(): + res.set_result(r) + despawn() + + dlg = Dialog( + title=title, + body=body, + buttons=[ + Button("Yes",handler=lambda: handler(True)), + Button("No",handler=lambda: handler(False)) + ] + ) + + despawn = spawnDialog(dlg) + + return await res + def HButtonSelect(buttonArgs): kb = KeyBindings() @@ -347,10 +373,12 @@ async def mainScreen(): if guestLogin: name = "[guest]" + admin = False else: stat = await server.status() if stat["type"]=="error" or not stat["response"]["login"]: return name = stat["response"]["name"] + admin = stat["response"]["admin"] def restart(): get_app().exit(result="restart") @@ -596,6 +624,161 @@ async def mainScreen(): ("","Text encoding: "),("bold",encoding),("","\n") ] + # ------------------------------------------------------------ + # Settings -> Admin panel -> Ban user + # ------------------------------------------------------------ + + async def banuserconfirm(name,duration,motive): + res = await server.getuserinfo(name) + if res["type"]=="error": + return await spawnSimplePopup("Error",f"{errorTxt(res)}\nCould not get information about the user.") + info = res["response"] + create_time = datetime.datetime.fromtimestamp(info["created"]) + motivestr = f"\"{motive}\"" if len(motive)>0 else "N/A" + conf = await spawnYesNoPopup("Confirm Ban?",Window(FormattedTextControl([ + ("","You are about to ban "), + ("bold",info["name"]), + (""," for "), + ("bold",({0: "an indefinite amount of time",24*60*60: "24 hours",3*24*60*60: "3 days",7*24*60*60: "7 days"})[duration]), + ("",".\n\nMotive: "),("bold",motivestr),("","\n"), + ("","\nUser information:\n"), + ("","- Internal ID: "),("bold",str(info["id"])),("","\n"), + ("","- Created: "),("bold",humanize.naturaldate(create_time)),("",f" ({create_time})\n") + ],focusable=False),dont_extend_width=True,dont_extend_height=True)) + if conf: + res = await server.banuser(info["id"],duration,motive) + if res["type"]=="error": + await spawnSimplePopup("Error",f"{errorTxt(res)}\nCould not ban this user.") + + def banuser(): + def dlgclose(): + despawn() + get_app().create_background_task(banuserconfirm(usertextfield.text,timeselect.current_value,motivetextfield.text)) + + def nameaccept(buf: Buffer): + get_app().layout.focus(motivetextfield) + return True + + def motiveaccept(buf: Buffer): + get_app().layout.focus(timeselect) + return True + + usernameValidator = Validator.from_callable(isValidUsername, error_message='Invalid username') + + ok_button = Button(text="OK", handler=dlgclose, width=16) + cancel_button = Button(text="Cancel", handler=lambda: despawn(), width=16) + + usertextfield = TextArea( + text="", + multiline=False, + password=False, + completer=None, + validator=usernameValidator, + accept_handler=nameaccept, + ) + + motivetextfield = TextArea( + text="", + multiline=False, + password=False, + completer=None, + validator=None, + accept_handler=motiveaccept, + ) + + timeselect = RadioList([(24*60*60,"24 hours"),(3*24*60*60,"3 days"),(7*24*60*60,"7 days"),(0,"Permanent")],default=intset["color"]) + + dlg = Dialog( + title="Ban user", + body=HSplit( + [ + Label(text="Username:", dont_extend_height=True), + usertextfield, + Label(text="Motive:", dont_extend_height=True), + motivetextfield, + Label(text="Time:", dont_extend_height=True), + timeselect, + ValidationToolbar(), + ], + padding=D(preferred=1, max=1), + width=70 + ), + buttons=[ok_button, cancel_button], + with_background=True, + ) + + despawn = spawnDialog(dlg) + + # ------------------------------------------------------------ + # Settings -> Admin panel -> Unban user + # ------------------------------------------------------------ + + async def unbanuserconfirm(name): + res = await server.getuserinfo(name) + if res["type"]=="error": + return await spawnSimplePopup("Error",f"{errorTxt(res)}\nCould not get information about the user.") + info = res["response"] + if info["banned"]==False: + return await spawnSimplePopup("Error",f"The user {name} is not banned.") + create_time = datetime.datetime.fromtimestamp(info["created"]) + ban_date = datetime.datetime.fromtimestamp(info["ban_details"]["ban_date"]) + expire_date = datetime.datetime.fromtimestamp(info["ban_details"]["expire_date"]) if info["ban_details"]["expire_date"]>0 else "Indefinite" + motive = f"\"{info["ban_details"]["motive"]}\"" if ("motive" in info["ban_details"]) and (len(info["ban_details"]["motive"])>0) else "N/A" + conf = await spawnYesNoPopup("Confirm Ban?",Window(FormattedTextControl([ + ("","You are about to unban "), + ("bold",info["name"]), + ("",".\n\nUser information:\n"), + ("","- Internal ID: "),("bold",str(info["id"])),("","\n"), + ("","- Created: "),("bold",humanize.naturaldate(create_time)),("",f" ({create_time})\n"), + ("","\nBan information:\n"), + ("","- Motive: "),("bold",motive),("","\n"), + ("","- Banned on: "),("bold",humanize.naturaldate(ban_date)),("",f" ({ban_date})\n"), + ("","- Ban expires on: "),("bold",humanize.naturaldate(expire_date)),("",f" ({expire_date})\n"), + ],focusable=False),dont_extend_width=True,dont_extend_height=True)) + if conf: + res = await server.unbanuser(info["id"]) + if res["type"]=="error": + await spawnSimplePopup("Error",f"{errorTxt(res)}\nCould not unban this user.") + + def unbanuser(): + def dlgclose(): + despawn() + get_app().create_background_task(unbanuserconfirm(usertextfield.text)) + + def nameaccept(buf: Buffer): + dlgclose() + return True + + usernameValidator = Validator.from_callable(isValidUsername, error_message='Invalid username') + + ok_button = Button(text="OK", handler=dlgclose, width=16) + cancel_button = Button(text="Cancel", handler=lambda: despawn(), width=16) + + usertextfield = TextArea( + text="", + multiline=False, + password=False, + completer=None, + validator=usernameValidator, + accept_handler=nameaccept, + ) + + dlg = Dialog( + title="Unban user", + body=HSplit( + [ + Label(text="Username:"), + usertextfield, + ValidationToolbar(), + ], + padding=D(preferred=1, max=1), + ), + buttons=[ok_button, cancel_button], + with_background=True, + ) + + despawn = spawnDialog(dlg) + # ------------------------------------------------------------ # Settings Menu # ------------------------------------------------------------ @@ -615,10 +798,12 @@ async def mainScreen(): Box(HButtonSelect([("Change color depth",changecolor),("Change text encoding",changeenc),("Apply changes",restart)]),padding=0), Label("\n\nYou will need to restart the interface or relog to apply any new changes."), ]) - ],[ + ] + ([ HButtonSelect([("Ban user",banuser),("Unban user",unbanuser)]) ] if admin else []), + [ + None, None, None - ],["Account","Interface"]) + ],["Account","Interface"]+(["Admin panel"] if admin else [])) # ------------------------------------------------------------ # Tab Layout