sourceserver.info

import es, os, popuplib, random, msglib, socket, re, xdrlib, string, threading, usermsg, gamethread, playerlib

#################
# CONFIGURATION #
#################

# Say commands which trigger PopupRedirect.
commands = ["!servers"]

# Advertise the PopupRedirect server command?
# Type <COMMAND> to see a list of our servers. (Will choose a random say command from the commands list.)
adverts = es.ServerVar("pr_adverts", 1, "Advertise the PopupRedirect server command?")

# How often to display adverts. Will display every X minutes until unloaded.
advertdelay = es.ServerVar("pr_advertdelay", 15, "How often to display adverts. In minutes.")

#################
# END OF CONFIG #
#################

serverlist = []
servernames = {}
players = {}
YN = {0: "No", 1: "Yes"}
es.ServerVar("popupredirect_ver", "1.0", "The version of PopupRedirect this server is running.").makepublic()

def player_connect(event_var):
showServers(event_var["userid"])

def load():
if not es.exists("command", "pr_addserver"): es.regcmd("pr_addserver", "popupredirect/addServer", "Add a server to PopupRedirect.")
if not es.exists("command", "pr_delserver"): es.regcmd("pr_delserver", "popupredirect/delServer", "Delete a server from PopupRedirect.")
if not es.exists("command", "pr_listservers"): es.regcmd("pr_listservers", "popupredirect/listServers", "List all PopupRedirect servers.")
loadServers()
displayAdvert()

def unload():
gamethread.cancelDelayed("advert")

def player_say(event_var):
if event_var["text"] in commands: showServers(event_var["userid"])

def displayAdvert():
if adverts == 1: es.msg("#multi", "#green[PopupRedirect]#default Type #lightgreen%s #defaultto see a list of our servers." % str(random.choice(commands)))
gamethread.delayedname(int(round(float(advertdelay) * 60)), "adverts", displayAdvert)

def loadServers():
global serverlist
serverlist = []
if not os.path.exists(es.getAddonPath("popupredirect") + "/servers.txt"): open(es.getAddonPath("popupredirect") + "/servers.txt", "w").close()
if os.path.exists(es.getAddonPath("popupredirect") + "/servers.txt"):
servers = open(es.getAddonPath("popupredirect") + "/servers.txt", "r").read()
for server in servers.split("\n"):
if not server.strip() == "" and not server.startswith("//"):
serverlist.append(server)
ServerInfo(server.split(":")[0], int(server.split(":")[1]), None, 2).start()

def addServer():
if es.getargc() > 1:
address = es.getargs()
if len(address.split(":")) > 1:
ip = address.split(":")[0]
port = address.split(":")[1]
address = ip + ":" + port
if checkServer(address):
serverlist = []
exists = 0
if os.path.exists(es.getAddonPath("popupredirect") + "/servers.txt"):
servers = open(es.getAddonPath("popupredirect") + "/servers.txt", "r").read()
for server in servers.split("\n"):
if not server.strip() == "" and not server.startswith("//"): serverlist.append(server)
if server == address: exists = 1
if exists == 0:
servers = open(es.getAddonPath("popupredirect") + "/servers.txt", "w")
text = "// Place server IP's that you want to show up in PopupRedirect here.\n// Usage: ip:port\n\n"
serverlist.append(address)
for server in serverlist: text += "%s\n" % server
servers.write(text)
servers.close()
es.dbgmsg(0, "%s has been added to the serverlist!" % address)
loadServers()
else: es.dbgmsg(0, "%s is already in the serverlist!" % address)
else: es.dbgmsg(0, "Invalid IP address!")
else: es.dbgmsg(0, "Syntax: addServer ip:port")
else: es.dbgmsg(0, "Syntax: addServer ip:port")

def delServer():
if es.getargc() > 1:
address = es.getargs()
if len(address.split(":")) > 1:
ip = address.split(":")[0]
port = address.split(":")[1]
address = ip + ":" + port
if checkServer(address):
serverlist = []
exists = 0
if os.path.exists(es.getAddonPath("popupredirect") + "/servers.txt"):
servers = open(es.getAddonPath("popupredirect") + "/servers.txt", "r").read()
for server in servers.split("\n"):
if not server.strip() == "" and not server.startswith("//") and not server == address: serverlist.append(server)
if server == address: exists = 1
if exists == 1:
servers = open(es.getAddonPath("popupredirect") + "/servers.txt", "w")
text = "// Place server IP's that you want to show up in PopupRedirect here.\n// Usage: ip:port\n\n"
for server in serverlist: text += "%s\n" % server
servers.write(text)
servers.close()
es.dbgmsg(0, "%s has been removed from the serverlist!" % address)
loadServers()
else: es.dbgmsg(0, "%s is not in the serverlist!" % address)
else: es.dbgmsg(0, "Invalid IP address!")
else: es.dbgmsg(0, "Syntax: delServer ip:port")
else: es.dbgmsg(0, "Syntax: delServer ip:port")

def listServers():
es.dbgmsg(0, "-------------------------")
es.dbgmsg(0, "PopupRedirect Server List")
es.dbgmsg(0, "-------------------------")
for server in serverlist:
name = "Unknown"
if server in servernames:
name = servernames[server]
if len(name) > 43: name = name[:40] + "..."
es.dbgmsg(0, "%-*s - %s" % (43, name, server))
es.dbgmsg(0, "-------------------------")

def checkServer(address):
if not len(address.split(":")) == 2: return False
ip = address.split(":")[0]
port = address.split(":")[1]
for x in ip.split("."):
try: int(x)
except: return False
try: int(port)
except: return False
return True

def showServers(userid):
p_servers = popuplib.easymenu("servers", "_popup_choice", showServerInfo)
p_servers.settitle("-PopupRedirect-\nSelect a server:")
for server in serverlist:
name = server
if server in servernames:
name = servernames[server]
if len(name) > 43: name = name[:40] + "..."
p_servers.addoption(server, name)
p_servers.send(userid)

def showServerInfo(userid, choice, popup):
global players
ip, port = choice.split(":")[0], int(choice.split(":")[1])
players[userid] = ServerInfo(ip, port, userid)
players[userid].start()

class ServerInfo(threading.Thread):
def __init__(self, host, port, userid=None, type=1):
self.ip, self.port, self.userid, self.type = socket.gethostbyname(host), port, userid, type
self.udpsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.tcpsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if type == 1: gamethread.delayedname(5.0, "timeout_%s" % self.userid, self.timeout)
try: self.udpsock.connect((self.ip, self.port))
except:
if type == 1: usermsg.hudhint(userid, "[PopupRedirect] Could not connect to server!")
threading.Thread.__init__(self)

def run(self):
if self.type == 1:
usermsg.hudhint(self.userid, "[PopupRedirect] Retrieving server information...")
self.showServerInfo()
else:
global servernames
servernames[self.ip + ":" + str(self.port)] = self.getServerName()

def getResponse(self, query):
self.udpsock.send("\xFF\xFF\xFF\xFF" + query)
data = self.udpsock.recv(4096)
return data[4:]

def getDetails(self):
data = self.getResponse("TSource Engine Query\x00")[1:]
detaildict = {}
detaildict["hl_version"] = 2
detaildict["protocol_version"], data = read_byte(data)
detaildict["server_name"], data = read_string(data)
detaildict["current_map"], data = read_string(data)
detaildict["game_directory"], data = read_string(data)
detaildict["game_description"], data = read_string(data)
aid1, data = read_byte(data)
aid2, data = read_byte(data)
detaildict["app_id"] = (aid1 * 0x100) + aid2
detaildict["current_playercount"], data = read_byte(data)
detaildict["max_players"], data = read_byte(data)
detaildict["current_botcount"], data = read_byte(data)
ded, data = read_char(data)
if ded == "d": detaildict['server_type'] = "Dedicated"
else: detaildict['server_type'] = "Listen"
os, data = read_char(data)
if os == "w": detaildict['server_os'] = "Windows"
else: detaildict['server_os'] = "Linux"
pworded, data = read_byte(data)
detaildict["passworded"] = bool(int(pworded))
secured, data = read_byte(data)
detaildict["secure"] = bool(int(secured))
detaildict["exe_version"], data = read_string(data)
return detaildict

def showServerInfo(self):
info = popuplib.create("info")
info.addline("-PopupRedirect-")
info.addline("---------------")
info.addline(self.getServerName())
gamethread.cancelDelayed("timeout_%s" % self.userid)
info.addline("---------------")
info.addline("IP: %s" % self.ip + ":" + str(self.port))
info.addline("Map: %s" % self.getServerMap())
info.addline("Players: %s" % self.getServerPlayerCount())
info.addline("Bots: %s" % self.getServerBots())
info.addline("Passworded: %s" % self.getServerPassword())
info.addline("Secure: %s" % self.getServerSecure())
info.addline("---------------")
info.addline("->1. Connect")
info.addline("->2. Players")
info.addline("---------------")
info.addline("->8. Back")
info.addline("0. Close")
info.menuselect = self.ConnectToServer
info.send(self.userid)
usermsg.hudhint(self.userid, "[PopupRedirect] Retrieved server information!")

def showPlayers(self):
try:
p_players = popuplib.easymenu("players", "_popup_choice", self.showPlayerInfo)
p_players.settitle("-PopupRedirect-")
players = self.getServerPlayers()
items = players.items()
backitems = [[v[1], v[0]] for v in items]
backitems.sort()
playerlist = []
for i in range(0, len(backitems)):
playerlist.append("%s:%s" % (backitems[1], players[backitems[i][1]]))
playerlist.reverse()
for player in playerlist:
p_players.addoption("", "%s - %s Frags" % (player.split(":")[0], player.split(":")[1]))
p_players.send(self.userid)
except: usermsg.hudhint(self.userid, "[PopupRedirect] Could not retrieve player list!")

def showPlayerInfo(self, userid, player, popup): pass

def ConnectToServer(self, userid, choice, popup):
if choice == 1: msglib.VguiDialog(title=self.ip + ":" + str(self.port), time=5, mode=msglib.VguiMode.ASKCONNECT).send(self.userid)
elif choice == 2: self.showPlayers()
elif choice == 8: showServers(self.userid)

def getServerName(self):
return self.getDetails()["server_name"]

def getServerMap(self):
return self.getDetails()["current_map"]

def getServerPlayerCount(self):
det = self.getDetails()
players = "%s/%s" % (str(det["current_playercount"]), str(det["max_players"]))
return players

def getServerBots(self):
return self.getDetails()["current_botcount"]

def getServerPassword(self):
return YN[int(self.getDetails()["passworded"])]

def getServerSecure(self):
return YN[int(self.getDetails()["secure"])]

def getServerPlayers(self):
challenge = self.getChallenge()
players = self.getResponse("U" + hldspack_int(challenge))
data = players[1:]
playerlist = {}
playercount, data = read_byte(data)
playercount = int(playercount)
while len(data) != 0:
currentplayer = {}
cn, data = read_byte(data)
currentplayer['index'] = int(cn)
currentplayer['name'], data = read_string(data)
currentplayer['frags'], data = read_int(data)
currentplayer['time_on'], data = read_float(data)
playerlist[currentplayer['name']] = currentplayer['frags']
return playerlist

def getChallenge(self):
challenge = self.getResponse("W")
data = challenge[1:]
challenge, data = read_int(data)
return challenge

def timeout(self):
usermsg.hudhint(self.userid, "[PopupRedirect] Unable to connect to server!")
self.udpsock.close()
self.tcpsock.close()

def hldsunpack_int(data):
s = ""
for c in data: s = c + s
p = xdrlib.Unpacker(s)
return p.unpack_int()

def hldsunpack_float(data):
s = ""
for c in data: s = c + s
p = xdrlib.Unpacker(s)
return p.unpack_float()

def hldspack_int(integer):
s = ""
p = xdrlib.Packer()
p.pack_int(integer)
data = p.get_buffer()
for c in data: s = c + s
return s

def read_byte(data):
return (ord(data[0]), data[1:])

def read_char(data):
return (str(data[0]), data[1:])

def read_string(data):
s = ''
i = 0
while 1:
try:
if str(data[i]) != '\x00':
s = s + str(data[i])
i += 1
else: break
except: pass
return (s, data[i+1:])

def read_int(data):
ret = hldsunpack_int(data[0:4])
return (ret, data[4:])

def read_float(data):
ret = hldsunpack_float(data[0:4])
return (ret, data[4:])