fstream/server.py

#!/usr/bin/env python3
import sys

oldexcepthook = sys.excepthook
def newexcepthook(type,value,traceback):
	oldexcepthook(type,value,traceback)
	#input("Press ENTER to quit.")
sys.excepthook = newexcepthook

import os
p = os.path.join
pUp = os.path.dirname
s = False
if getattr(sys, 'frozen', False) and hasattr(sys, '_MEIPASS'):
	s = os.path.realpath(sys.executable)
else:
	s = os.path.realpath(__file__)
sp = pUp(s)

import threading
import queue
import socket
import configparser

debug = True
bufferSize = 8096 # 8 KiB
timeout = 15
maxClients = 100
maxClientsPerIP = 10
maxBroadcastsPerIP = 3
maxAccumulatedData = 10485760 # 10 MiB
addresses = [
	("127.0.0.1",61920)
]
userPath = p(sp,"users")

clients = {}
clientsID = 0
clientsLock = threading.Lock()

pendingConnections = queue.Queue()

def move (y, x):
    print("\033[%d;%dH" % (y, x))

def listToCommand(lst):
	cmd = ""
	for arg in lst:
		arg = arg.replace("\\","\\\\")
		arg = arg.replace(",","\\,")
		cmd += arg + ","
	
	return cmd[:-1]

def commandToList(cmd):
	args = []
	cArg = ""
	escape = False
	for letter in cmd:
		if escape == True:
			cArg += letter
			escape = False
			continue
		
		if letter == "\\":
			escape = True
			continue
		
		if letter == ",":
			if cArg == "": continue
			args.append(cArg)
			cArg = ""
			continue
		
		cArg += letter
	
	args.append(cArg)
	
	return args

def makePayload(lst):
	cmdText = listToCommand(lst)
	cmdBytes = cmdText.encode("utf-8")
	cmdBytes += b" " * (1024 - len(cmdBytes))
	return cmdBytes

class socketThread(threading.Thread):
	def __init__(self,ip):
		threading.Thread.__init__(self)
		self.ip = ip
		self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		self.socket.bind(self.ip)
	
	def run(self):
		self.socket.listen(32766)
		while True:
			connection = False
			try:
				connection,address = self.socket.accept()
				pendingConnections.put((connection,address))
			except Exception as e:
				try:
					connection.close()
				except:
					pass
				
				print(e)

class clientThread(threading.Thread):
	def __init__(self,clientID,env):
		threading.Thread.__init__(self)
		self.clientID = clientID
		self.env = env
		self.queue = queue.Queue()
	
	def childRun(self):
		if self.env["command"] == "watch":
			self.env["connection"].sendall(makePayload([str(self.env["bsize"])]))
			while True:
				data = self.queue.get(timeout=timeout)
				while self.queue.qsize() > 0:
					try:
						data += self.queue.get(False)
					except Queue.Empty:
						pass
						
				self.env["connection"].sendall(data)
		
		if self.env["command"] == "broadcast":
			self.env["connection"].sendall(makePayload(["dummy"]))
			while True:
				data = self.env["connection"].recv(self.env["bsize"])
				if data == b"": return
				with clientsLock:
					for clientID in clients:
						try:
							if clients[clientID]["command"] != "watch": continue
							if clients[clientID]["client"] != str(self.clientID): continue
							clients[clientID]["thread"].queue.put(data)
						except:
							pass
	
	def run(self):
		try:
			self.childRun()
		except Exception as e:
			try:
				print(e)
			except:
				print("Could not print exception.")
		
		try:
			self.env["connection"].close()
		except:
			pass
		
		try:
			with clientsLock:
				del clients[str(self.clientID)]
		except:
			pass

def handleConnection(connection,address):
	global clientsID
	args = commandToList(connection.recv(1024).decode("utf-8").rstrip(" "))
	command = args[0]
	if command == "watch":
		user = args[1]
		room = args[2]
		passw = args[3]
		
		watchID = False
		with clientsLock:
			for clientID in clients:
				if clients[clientID]["command"] != "broadcast": continue
				if clients[clientID]["user"] != user: continue
				if clients[clientID]["room"] != room: continue
				if clients[clientID]["passw"] != passw: continue
				watchID = clientID
				break
			
			if watchID == False:
				connection.close()
				return
			
			env = {
				"command":    "watch",
				"client":     watchID,
				"connection": connection,
				"address":    address,
				"bsize":      clients[watchID]["bsize"]
			}
			
			clientsID += 1
			thr = clientThread(clientsID,env.copy())
			env["thread"] = thr
			clients[str(clientsID)] = env
			thr.start()
		return
	
	if command == "broadcast":
		bsize = int(args[1])
		user = args[2]
		userPassw = args[3]
		room = args[4]
		roomPassw = args[5]
		
		config = configparser.ConfigParser()
		config.read(p(userPath,user+ ".ini"))
		if config["DEFAULT"]["password"] != userPassw:
			connection.close()
			return
		
		env = {
			"command":    "broadcast",
			"user":       user,
			"room":       room,
			"passw":      roomPassw,
			"bsize":      bsize,
			"connection": connection,
			"address":    address
		}
		
		with clientsLock:
			clientsID += 1
			thr = clientThread(clientsID,env.copy())
			env["thread"] = thr
			clients[str(clientsID)] = env
			thr.start()
		return
	
	connection.close()

class debugThread(threading.Thread):
	def __init__(self):
		threading.Thread.__init__(self)
		global time
		import time
	
	def run(self):
		while True:
			clientCount = 0
			accumDataCount = 0
			threadCount = threading.active_count()
			with clientsLock:
				clientCount = len(clients)
				for clientID in clients:
					accumDataCount += (clients[clientID]["thread"].queue.qsize() * clients[clientID]["bsize"]) / 1000000
			
			#move(0,0)
			print("   ---   ")
			print("   Threads: " +str(threadCount)+ "   ")
			print("   Clients: " +str(clientCount)+ "   ")
			print("   Accumulated Data: " +str(accumDataCount)+ "MB   ")
			time.sleep(1)

def main():
	for address in addresses:
		print("Opening TCP socket " +address[0]+ ":" +str(address[1])+ "...")
		thr = socketThread(address)
		thr.start()
	
	print("Serving!")
	if debug == True:
		thr = debugThread()
		thr.start()
	
	while True:
		connection,address = pendingConnections.get()
		try:
			handleConnection(connection,address)
		except Exception as e:
			try:
				connection.close()
			except:
				pass
			
			try:
				print(e)
			except:
				print("Could not print exception.")

if __name__ == '__main__':
	main()