LinuxGSM/lgsm/functions/gsquery.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# Game Server Query
# Author: Anonymous & Daniel Gibbs
# Website: https://gameservermanagers.com
# Description: Handles querying of .

import optparse
import socket
import sys


class GameServer:
    def __init__(self, options, arguments):
        self.option = options
        self.argument = arguments
        #
        self.server_response_timeout = 5
        self.default_buffer_length = 1024
        #
        if self.option.engine == 'avalanche':
            self.query_prompt_string = b'\xFE\xFD\x09\x10\x20\x30\x40'
        elif self.option.engine == 'goldsource':
            self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0'
        elif self.option.engine == 'idtech3':
            self.query_prompt_string = b'\xff\xff\xff\xffgetstatus'
        elif self.option.engine == 'quakelive':
            self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0'
        elif self.option.engine == 'realvirtuality':
            self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0'
        elif self.option.engine == 'refractor':
            self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0'
        elif self.option.engine == 'source':
            self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0'
        elif self.option.engine == 'spark':
            self.query_prompt_string = b'\xFF\xFF\xFF\xFFTSource Engine Query\0'
        elif self.option.engine == 'unity3d':
            self.query_prompt_string = '\xFF\xFF\xFF\xFFTSource Engine Query\0'
        elif self.option.engine == 'unreal':
            self.query_prompt_string = b'\x5C\x69\x6E\x66\x6F\x5C'
        elif self.option.engine == 'unreal2':
            self.query_prompt_string = b'\x79\x00\x00\x00\x00'
        self.connected = False
        self.response = None
        self.sanity_checks()

    def fatal_error(self, error_message, error_code=1):
        sys.stderr.write('ERROR: ' + str(error_message) + '\n')
        sys.exit(error_code)

    def exit_success(self, success_message=''):
        sys.stdout.write('OK: ' + str(success_message) + '\n')
        sys.exit(0)

    def responding(self):
        # Connect.
        connection = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        connection.settimeout(self.server_response_timeout)
        try:
            self.connected = connection.connect((self.option.address, int(self.option.port)))
        except socket.timeout:
            self.fatal_error('Request timed out', 1)
        except:
            self.fatal_error('Unable to connect', 1)
        # Send.
        connection.send(self.query_prompt_string)
        # Receive.
        try:
            self.response = connection.recv(self.default_buffer_length)
        except socket.error:
            self.fatal_error('Unable to receive', 2)
        connection.close()
        # Response.
        if self.response is None:
            self.fatal_error('No response', 3)
        if len(self.response) < 10:
            sys.exit('Short response.', 3)
        else:
            self.exit_success(str(self.response))

    def sanity_checks(self):
        if not self.option.address:
            self.fatal_error('No IPv4 address supplied.', 4)
        if not self.option.port:
            self.fatal_error('No port supplied.', 4)


if __name__ == '__main__':
    parser = optparse.OptionParser(
        usage='usage: python %prog [options]',
        version='%prog 0.0.1'
    )
    parser.add_option(
        '-a', '--address',
        action='store',
        dest='address',
        default=False,
        help='The IPv4 address of the server.'
    )
    parser.add_option(
        '-p', '--port',
        action='store',
        dest='port',
        default=False,
        help='The IPv4 port of the server.'
    )
    parser.add_option(
        '-e', '--engine',
        action='store',
        dest='engine',
        default=False,
        help='Engine type: avalanche, goldsource, idtech3, realvirtuality, quakelive, refractor, spark, source, unity3d, unreal, unreal2.'
    )
    parser.add_option(
        '-v', '--verbose',
        action='store_true',
        dest='verbose',
        default=False,
        help='Display verbose output.'
    )
    parser.add_option(
        '-d', '--debug',
        action='store_true',
        dest='debug',
        default=False,
        help='Display debugging output.'
    )
    options, arguments = parser.parse_args()
    #
    server = GameServer(options, arguments)
    server.responding()