gandon_rejector/scan_from_list.py at main · FWGS/gandon_rejector · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
#!/usr/bin/env python3
import socket
import random
import sys
import traceback
import logging
import os
import datetime
import requests
import hashlib
import re
from optparse import OptionParser
from struct import pack
from time import time, strftime, localtime
from utils import tobuf, toutf, tostr, dict_to_info, info_to_dict
from alarm import send_alarm
import config

COLORSTRING_RE = re.compile(r'(\^[0-9]|[|])')
PACKET_HEADER = b'\xff\xff\xff\xff'

def get_socket(options, timeout=1):
	sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM | socket.SOCK_NONBLOCK)
	sock.settimeout(timeout)
	if options.netdev:
		sock.setsockopt(socket.SOL_SOCKET, socket.SO_BINDTODEVICE, tobuf(options.netdev))
	sock.bind(('0.0.0.0', 0))
	return sock

def outofbound_print(sock, addr, message):
	p = tobuf('\xff\xff\xff\xff' + message)
	sock.sendto(p, addr)

def collect_servers(sock, path):
	servers = []
	with open(path, 'r') as f:
		lines = f.readlines()
		for i in lines:
			if i[0] == '#':
				continue

			ip = i.split(':')[0]
			port = int(i.split(':')[1])
			servers += [(ip, port)]
	return servers

def validate_server_info(sock):
	while True:
		try:
			(data, addr) = sock.recvfrom(2048)
		except BlockingIOError:
			continue
		except (socket.timeout, TimeoutError):
			continue

		if data[0:4] != PACKET_HEADER:
			print('Unhandled packet %s from %s:%i' % (data, addr[0], addr[1]))
			continue
		data = data.strip(PACKET_HEADER)

		if data.startswith(b'info\n\\'): # infostring begin
			data = data.strip(b'info\n\\')
			info = info_to_dict(tostr(data))

			print(addr, ' ', data)
		else:
			print('broken ', addr, ' ', data)

def probe_servers_from_multiple_sockets(options, server_list):
	bad_servers = []
	ok_servers = []
	sock = get_socket(options)

	for server in server_list:
		outofbound_print(sock, server, 'info %d' % options.protocol)

	validate_server_info(sock)

def main(options):
	sock = get_socket(options, 5)

	if options.unban and not options.probe:
		print('--unban must be used with --probe-server')
		return

	server_list = collect_servers(sock, options.listfile)

	if options.list:
		random.shuffle(server_list)
		for i in server_list:
			print("%s:%d" % (i[0], i[1]))
		return

	print('Found servers: %d' % (len(server_list)))

	bad_servers = probe_servers_from_multiple_sockets(options, server_list)

if __name__ == "__main__":
	parser = OptionParser()
	parser.add_option('--probe-server', action='store', dest='probe', default=None)
	parser.add_option('-i', '--input-list', action='store', dest='listfile', default=None)
	parser.add_option('-g', '--gamedir', action='store', dest='gamedir', default='valve')
	parser.add_option('-p', '--protocol', action='store', dest='protocol', default=48, type='int')
	parser.add_option('-d', '--netdev', action='store', dest='netdev', default=None)
	parser.add_option('-l', '--list-ips', action='store_true', dest='list', default=False)
	parser.add_option('--dry-run', action='store_true', dest='dry', default=False)
	parser.add_option('--unban', action='store_true', dest='unban', default=False)

	(options, args) = parser.parse_args()
	if not options.list:
		print('=============================================')
	main(options)
	if not options.list:
		print('ENDENDENDENDENDENDENDENDENDENDENDENDENDENDEND')