1
0
Fork 0
mirror of https://github.com/de2tla2f/pymaster.git synced 2024-11-22 02:51:28 +00:00

refactor: delete some shitcode

This commit is contained in:
def 2022-10-23 16:12:35 +04:00
parent 4fa50a3bd0
commit d8c8fe9c54
2 changed files with 442 additions and 227 deletions

351
pymaster.py Executable file → Normal file
View file

@ -12,192 +12,241 @@ from time import time
from server_entry import ServerEntry from server_entry import ServerEntry
from protocol import MasterProtocol from protocol import MasterProtocol
LOG_FILENAME = 'pymaster.log' LOG_FILENAME = "pymaster.log"
MAX_SERVERS_FOR_IP = 14 MAX_SERVERS_FOR_IP = 14
def logPrint( msg ):
logging.debug( msg )
class PyMaster: class PyMaster:
def __init__(self, ip, port): def __init__(self, ip, port):
self.serverList = [] self.serverList = []
self.sock = socket.socket( socket.AF_INET, socket.SOCK_DGRAM ) self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.bind( (ip, port) ) self.sock.bind((ip, port))
logPrint("Welcome to PyMaster!") logging.debug("Welcome to PyMaster!")
logPrint("I ask you again, are you my master?") logging.debug("I ask you again, are you my master? @-@")
logPrint("Running on %s:%d" % (ip, port)) logging.debug("Running on %s:%d" % (ip, port))
def serverLoop(self):
data, addr = self.sock.recvfrom(1024)
data = data.decode('latin_1')
if( data[0] == MasterProtocol.clientQuery ): def server_loop(self):
self.clientQuery(data, addr) data, addr = self.sock.recvfrom(1024)
elif( data[0] == MasterProtocol.challengeRequest ): data = data.decode("latin_1")
self.sendChallengeToServer(data, addr)
elif( data[0] == MasterProtocol.addServer ):
self.addServerToList(data, addr)
elif( data[0] == MasterProtocol.removeServer ):
self.removeServerFromList(data, addr)
else:
logPrint("Unknown message: {0} from {1}:{2}".format(data, addr[0], addr[1]))
def clientQuery(self, data, addr): match data[0]:
region = data[1] # UNUSED case MasterProtocol.clientQuery:
data = data.strip('1' + region) self.client_query(data, addr)
try: case MasterProtocol.challengeRequest:
query = data.split('\0') self.send_challenge_to_server(data, addr)
except ValueError: case MasterProtocol.addServer:
logPrint(traceback.format_exc()) self.add_server_to_list(data, addr)
return case MasterProtocol.removeServer:
self.remove_server_from_list(data, addr)
case other:
logging.debug("Unknown message: {0} from {1}:{2}".format(data, addr[0], addr[1]))
queryAddr = query[0] # UNUSED
rawFilter = query[1]
# Remove first \ character def client_query(self, data, addr):
rawFilter = rawFilter.strip('\\') region = data[1] # UNUSED
split = rawFilter.split('\\') data = data.strip("1" + region)
try:
query = data.split("\0")
except ValueError:
logging.debug(traceback.format_exc())
return
# Use NoneType as undefined queryAddr = query[0] # UNUSED
gamedir = 'valve' # halflife, by default rawFilter = query[1]
clver = None
nat = 0
for i in range( 0, len(split), 2 ): # Remove first \ character
try: rawFilter = rawFilter.strip("\\")
key = split[i + 1] split = rawFilter.split("\\")
if( split[i] == 'gamedir' ):
gamedir = key.lower() # keep gamedir in lowercase
elif( split[i] == 'nat' ):
nat = int(key)
elif( split[i] == 'clver' ):
clver = key
else:
logPrint('Unhandled info string entry: {0}/{1}. Infostring was: {2}'.format(split[i], key, split))
except IndexError:
pass
if( clver == None ): # Probably an old vulnerable version # Use NoneType as undefined
self.fakeInfoForOldVersions( gamedir, addr ) gamedir = "valve" # halflife, by default
return clver = None
nat = 0
packet = MasterProtocol.queryPacketHeader for i in range(0, len(split), 2):
for i in self.serverList: try:
if( time() > i.die ): key = split[i + 1]
self.serverList.remove(i) if split[i] == "gamedir":
continue gamedir = key.lower() # keep gamedir in lowercase
elif split[i] == "nat":
nat = int(key)
elif split[i] == "clver":
clver = key
else:
logging.debug(
"Unhandled info string entry: {0}/{1}. Infostring was: {2}".format(
split[i], key, split
)
)
except IndexError:
pass
if( not i.check ): if clver is None: # Probably an old vulnerable version
continue self.fake_info_for_old_versions(gamedir, addr)
return
if( nat != i.nat ): packet = MasterProtocol.queryPacketHeader
continue for i in self.serverList:
if time() > i.die:
self.serverList.remove(i)
continue
if( gamedir != None ): if not i.check:
if( gamedir != i.gamedir ): continue
continue
if( nat ): if nat != i.nat:
reply = '\xff\xff\xff\xffc {0}:{1}'.format( addr[0], addr[1] ) continue
data = reply.encode( 'latin_1' )
# Tell server to send info reply
self.sock.sendto( data, i.addr )
# Use pregenerated address string if gamedir is not None and gamedir != i.gamedir:
packet += i.queryAddr continue
packet += b'\0\0\0\0\0\0' # Fill last IP:Port with \0
self.sock.sendto(packet, addr)
def fakeInfoForOldVersions(self, gamedir, addr): if nat:
def sendFakeInfo(sock, warnmsg, gamedir, addr): reply = "\xff\xff\xff\xffc {0}:{1}".format(addr[0], addr[1])
baseReply = b"\xff\xff\xff\xffinfo\n\host\\" + warnmsg.encode('utf-8') + b"\map\\update\dm\\0\\team\\0\coop\\0\\numcl\\32\maxcl\\32\\gamedir\\" + gamedir.encode('latin-1') + b"\\" data = reply.encode("latin_1")
sock.sendto(baseReply, addr) # Tell server to send info reply
self.sock.sendto(data, i.addr)
sendFakeInfo(self.sock, "This version is not", gamedir, addr) # Use pregenerated address string
sendFakeInfo(self.sock, "supported anymore", gamedir, addr) packet += i.queryAddr
sendFakeInfo(self.sock, "Please update Xash3DFWGS", gamedir, addr)
sendFakeInfo(self.sock, "From GooglePlay or GitHub", gamedir, addr)
sendFakeInfo(self.sock, "Эта версия", gamedir, addr)
sendFakeInfo(self.sock, "устарела", gamedir, addr)
sendFakeInfo(self.sock, "Обновите Xash3DFWGS c", gamedir, addr)
sendFakeInfo(self.sock, "GooglePlay или GitHub", gamedir, addr)
def removeServerFromList(self, data, addr): packet += b"\0\0\0\0\0\0" # Fill last IP:Port with \0
for i in self.serverList: self.sock.sendto(packet, addr)
if (i.addr == addr):
logPrint("Remove Server: from {0}:{1}".format(addr[0], addr[1]))
self.serverList.remove(i)
def sendChallengeToServer(self, data, addr):
logPrint("Challenge Request: from {0}:{1}".format(addr[0], addr[1]))
# At first, remove old server- data from list
#self.removeServerFromList(None, addr)
count = 0 def _send_fake_info(sock, warnmsg, gamedir, addr):
for i in self.serverList: baseReply = (
if ( i.addr[0] == addr[0] ): b"\xff\xff\xff\xffinfo\n\host\\"
if( i.addr[1] == addr[1] ): + warnmsg.encode("utf-8")
self.serverList.remove(i) + b"\map\\update\dm\\0\\team\\0\coop\\0\\numcl\\32\maxcl\\32\\gamedir\\"
else: + gamedir.encode("latin-1")
count += 1 + b"\\"
if( count > MAX_SERVERS_FOR_IP ): )
return sock.sendto(baseReply, addr)
# Generate a 32 bit challenge number
challenge = random.randint(0, 2**32-1)
# Add server to list def fake_info_for_old_versions(self, gamedir, addr):
self.serverList.append(ServerEntry(addr, challenge)) error_message = [
"This version is not",
"supported anymore",
"Please update Xash3DFWGS",
"From GooglePlay or GitHub",
"Эта версия",
"устарела",
"Обновите Xash3DFWGS c",
"GooglePlay или GitHub",
]
# And send him a challenge for string in error_message:
packet = MasterProtocol.challengePacketHeader _send_fake_info(self.sock, string, gamedir, addr)
packet += pack('I', challenge)
self.sock.sendto(packet, addr)
def addServerToList(self, data, addr):
logPrint("Add Server: from {0}:{1}".format(addr[0], addr[1]))
# Remove the header. Just for better parsing.
serverInfo = data.strip('\x30\x0a\x5c')
# Find a server with same address def remove_server_from_list(self, addr):
for serverEntry in self.serverList: for server in self.serverList:
if( serverEntry.addr == addr ): if server.addr == addr:
break logging.debug("Remove Server: from {0}:{1}".format(addr[0], addr[1]))
self.serverList.remove(server)
def send_challenge_to_server(self, addr):
logging.debug("Challenge Request: from {0}:{1}".format(addr[0], addr[1]))
# At first, remove old server- data from list
# self.removeServerFromList(None, addr)
count = 0
for i in self.serverList:
if i.addr[0] == addr[0]:
if i.addr[1] == addr[1]:
self.serverList.remove(i)
else:
count += 1
if count > MAX_SERVERS_FOR_IP:
return
challenge = random.randint(0, 2**32 - 1)
# Add server to list
self.serverList.append(ServerEntry(addr, challenge))
# And send him a challenge
packet = MasterProtocol.challengePacketHeader
packet += pack("I", challenge)
self.sock.sendto(packet, addr)
def add_server_to_list(self, data, addr):
logging.debug("Add Server: from {0}:{1}".format(addr[0], addr[1]))
# Remove the header. Just for better parsing.
serverInfo = data.strip("\x30\x0a\x5c")
# Find a server with same address
for serverEntry in self.serverList:
if serverEntry.addr == addr:
break
serverEntry.setInfoString(serverInfo)
serverEntry.setInfoString( serverInfo )
def spawn_pymaster(verbose, ip, port): def spawn_pymaster(verbose, ip, port):
if verbose: if verbose:
logging.getLogger().addHandler(logging.StreamHandler()) logging.getLogger().addHandler(logging.StreamHandler())
logging.getLogger().addHandler(logging.FileHandler(LOG_FILENAME)) logging.getLogger().addHandler(logging.FileHandler(LOG_FILENAME))
logging.getLogger().setLevel(logging.DEBUG) logging.getLogger().setLevel(logging.DEBUG)
masterMain = PyMaster(ip, port)
while True:
try:
masterMain.server_loop()
except Exception:
logging.debug(traceback.format_exc())
masterMain = PyMaster(ip, port)
while True:
try:
masterMain.serverLoop()
except Exception:
logPrint(traceback.format_exc())
pass
if __name__ == "__main__": if __name__ == "__main__":
parser = OptionParser()
parser.add_option('-i', '--ip', action='store', dest='ip', default='0.0.0.0',
help='ip to listen [default: %default]')
parser.add_option('-p', '--port', action='store', dest='port', type='int', default=27010,
help='port to listen [default: %default]')
parser.add_option('-d', '--daemonize', action='store_true', dest='daemonize', default=False,
help='run in background, argument is uid [default: %default]')
parser.add_option('-q', '--quiet', action='store_false', dest='verbose', default=True,
help='don\'t print to stdout [default: %default]')
(options, args) = parser.parse_args() parser = OptionParser()
parser.add_option(
"-i",
"--ip",
action="store",
dest="ip",
default="0.0.0.0",
help="ip to listen [default: %default]",
)
parser.add_option(
"-p",
"--port",
action="store",
dest="port",
type="int",
default=27010,
help="port to listen [default: %default]",
)
parser.add_option(
"-d",
"--daemonize",
action="store_true",
dest="daemonize",
default=False,
help="run in background, argument is uid [default: %default]",
)
parser.add_option(
"-q",
"--quiet",
action="store_false",
dest="verbose",
default=True,
help="don't print to stdout [default: %default]",
)
if options.daemonize != 0: (options, args) = parser.parse_args()
from daemon import pidfile, DaemonContext
with DaemonContext(stdout=sys.stdout, stderr=sys.stderr, working_directory=os.getcwd()) as context: if options.daemonize != 0:
spawn_pymaster(options.verbose, options.ip, options.port) from daemon import DaemonContext
else:
sys.exit(spawn_pymaster(options.verbose, options.ip, options.port)) with DaemonContext(
stdout=sys.stdout, stderr=sys.stderr, working_directory=os.getcwd()
) as context:
spawn_pymaster(options.verbose, options.ip, options.port)
else:
sys.exit(spawn_pymaster(options.verbose, options.ip, options.port))

View file

@ -1,83 +1,249 @@
from time import time #!/usr/bin/env python3
import socket
import random
import sys
import traceback
import logging
import os
from optparse import OptionParser
from struct import pack from struct import pack
from time import time
class ServerEntry: from server_entry import ServerEntry
challenge2 = 0 from protocol import MasterProtocol
gamedir = ''
protocol = 0 LOG_FILENAME = "pymaster.log"
players = 0 MAX_SERVERS_FOR_IP = 14
maxplayers = 0
bots = 0
gamemap = '' class PyMaster:
version = '0' def __init__(self, ip, port):
servtype = 'd' self.serverList = []
password = 0 self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
os = 'l' self.sock.bind((ip, port))
secure = 0
lan = 0 logging.debug("Welcome to PyMaster!")
region = 255 logging.debug("I ask you again, are you my master? @-@")
product = '' logging.debug("Running on %s:%d" % (ip, port))
nat = 0
def server_loop(self):
data, addr = self.sock.recvfrom(1024)
data = data.decode("latin_1")
match data[0]:
case MasterProtocol.clientQuery:
self.client_query(data, addr)
case MasterProtocol.challengeRequest:
self.send_challenge_to_server(data, addr)
case MasterProtocol.addServer:
self.add_server_to_list(data, addr)
case MasterProtocol.removeServer:
self.remove_server_from_list(data, addr)
case other:
logging.debug(
"Unknown message: {0} from {1}:{2}".format(data, addr[0], addr[1])
)
def client_query(self, data, addr):
region = data[1] # UNUSED
data = data.strip("1" + region)
try:
query = data.split("\0")
except ValueError:
logging.debug(traceback.format_exc())
return
queryAddr = query[0] # UNUSED
rawFilter = query[1]
# Remove first \ character
rawFilter = rawFilter.strip("\\")
split = rawFilter.split("\\")
# Use NoneType as undefined
gamedir = "valve" # halflife, by default
clver = None
nat = 0
def setInfoString(self, data):
infostring = data.replace('\n', '').replace('\r', '').replace('\0', '')
split = infostring.split('\\')
for i in range(0, len(split), 2): for i in range(0, len(split), 2):
try: try:
key = split[i + 1] key = split[i + 1]
if( split[i] == 'challenge' ): if split[i] == "gamedir":
self.challenge2 = int(key) gamedir = key.lower() # keep gamedir in lowercase
elif( split[i] == 'gamedir' ): elif split[i] == "nat":
self.gamedir = key.lower() # keep gamedir lowercase nat = int(key)
elif( split[i] == 'protocol' ): elif split[i] == "clver":
self.protocol = int(key) clver = key
elif( split[i] == 'players' ): else:
self.players = int(key) logging.debug(
elif( split[i] == 'max' ): "Unhandled info string entry: {0}/{1}. Infostring was: {2}".format(
self.maxplayers = int(key.split('.')[0]) split[i], key, split
elif( split[i] == 'bots' ): )
self.bots = int(key) )
elif( split[i] == 'map' ):
self.gamemap = key
elif( split[i] == 'version' ):
self.version = key
elif( split[i] == 'type' ):
self.servtype = key
elif( split[i] == 'password' ):
self.password = key
elif( split[i] == 'os' ):
self.os = key
elif( split[i] == 'secure' ):
self.secure = key
elif( split[i] == 'lan' ):
self.lan = key
elif( split[i] == 'region' ):
self.region = key
elif( split[i] == 'product' ):
self.product = key
elif( split[i] == 'nat' ):
self.nat = int(key)
except IndexError: except IndexError:
pass pass
self.check = self.challenge == self.challenge2
def __init__(self, addr, challenge): if clver is None: # Probably an old vulnerable version
# Address self.fake_info_for_old_versions(gamedir, addr)
self.addr = addr return
# Shortcuts for generating query packet = MasterProtocol.queryPacketHeader
self.queryAddr = b''
for i in addr[0].split('.'):
self.queryAddr += pack('!B', int(i))
self.queryAddr += pack('!H', int(addr[1]))
# Random number that server must return for i in self.serverList:
self.challenge = challenge if time() > i.die:
self.serverList.remove(i)
continue
if not i.check:
continue
if nat != i.nat:
continue
if gamedir is not None and gamedir != i.gamedir:
continue
if nat:
reply = "\xff\xff\xff\xffc {0}:{1}".format(addr[0], addr[1])
data = reply.encode("latin_1")
# Tell server to send info reply
self.sock.sendto(data, i.addr)
# This server is not checked # Use pregenerated address string
# So it will not get into queries packet += i.queryAddr
self.check = False
# Remove server after this time. packet += b"\0\0\0\0\0\0" # Fill last IP:Port with \0
# This maybe not instant self.sock.sendto(packet, addr)
self.die = time() + 600.0
def _send_fake_info(sock, warnmsg, gamedir, addr):
baseReply = (
b"\xff\xff\xff\xffinfo\n\host\\"
+ warnmsg.encode("utf-8")
+ b"\map\\update\dm\\0\\team\\0\coop\\0\\numcl\\32\maxcl\\32\\gamedir\\"
+ gamedir.encode("latin-1")
+ b"\\"
)
sock.sendto(baseReply, addr)
def fake_info_for_old_versions(self, gamedir, addr):
error_message = [
"This version is not",
"supported anymore",
"Please update Xash3DFWGS",
"From GooglePlay or GitHub",
"Эта версия",
"устарела",
"Обновите Xash3DFWGS c",
"GooglePlay или GitHub",
]
for string in error_message:
_send_fake_info(self.sock, string, gamedir, addr)
def remove_server_from_list(self, addr):
for server in self.serverList:
if server.addr == addr:
logging.debug("Remove Server: from {0}:{1}".format(addr[0], addr[1]))
self.serverList.remove(server)
def send_challenge_to_server(self, addr):
logging.debug("Challenge Request: from {0}:{1}".format(addr[0], addr[1]))
# At first, remove old server- data from list
# self.removeServerFromList(None, addr)
count = 0
for i in self.serverList:
if i.addr[0] == addr[0]:
if i.addr[1] == addr[1]:
self.serverList.remove(i)
else:
count += 1
if count > MAX_SERVERS_FOR_IP:
return
challenge = random.randint(0, 2**32 - 1)
# Add server to list
self.serverList.append(ServerEntry(addr, challenge))
# And send him a challenge
packet = MasterProtocol.challengePacketHeader
packet += pack("I", challenge)
self.sock.sendto(packet, addr)
def add_server_to_list(self, data, addr):
logging.debug("Add Server: from {0}:{1}".format(addr[0], addr[1]))
# Remove the header. Just for better parsing.
serverInfo = data.strip("\x30\x0a\x5c")
# Find a server with same address
for serverEntry in self.serverList:
if serverEntry.addr == addr:
break
serverEntry.setInfoString(serverInfo)
def spawn_pymaster(verbose, ip, port):
if verbose:
logging.getLogger().addHandler(logging.StreamHandler())
logging.getLogger().addHandler(logging.FileHandler(LOG_FILENAME))
logging.getLogger().setLevel(logging.DEBUG)
masterMain = PyMaster(ip, port)
while True:
try:
masterMain.server_loop()
except Exception:
logging.debug(traceback.format_exc())
if __name__ == "__main__":
parser = OptionParser()
parser.add_option(
"-i",
"--ip",
action="store",
dest="ip",
default="0.0.0.0",
help="ip to listen [default: %default]",
)
parser.add_option(
"-p",
"--port",
action="store",
dest="port",
type="int",
default=27010,
help="port to listen [default: %default]",
)
parser.add_option(
"-d",
"--daemonize",
action="store_true",
dest="daemonize",
default=False,
help="run in background, argument is uid [default: %default]",
)
parser.add_option(
"-q",
"--quiet",
action="store_false",
dest="verbose",
default=True,
help="don't print to stdout [default: %default]",
)
(options, args) = parser.parse_args()
if options.daemonize != 0:
from daemon import DaemonContext
with DaemonContext(
stdout=sys.stdout, stderr=sys.stderr, working_directory=os.getcwd()
) as context:
spawn_pymaster(options.verbose, options.ip, options.port)
else:
sys.exit(spawn_pymaster(options.verbose, options.ip, options.port))