Cerulean-Blue/StingNET
4
0
Fork 0
Code Issues 18 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

added admins and banning

Browse Source 
this took more time than i thought it would
This commit is contained in:
Ponali
2026-08-07 16:44:28 +02:00
parent 7d3a380cab
commit a83ad194eb
4 changed files with 299 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api.py
+21 -5
View File
@@ -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})
requirements.txt
+1
View File
@@ -1,2 +1,3 @@
humanize
pygments
prompt_toolkit
server/main.py
+90 -2
View File
@@ -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."))
telnet-basic/main.py
+187 -2
View File
@@ -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