Я тут подумал, что отпуск по-тихонечку подходит к концу и, возможно, многое из того, что я написал за 2 месяца просто пропадет в недрах моих ноутбуков навсегда... Это было интересно и познавательно, но возможно кому-то пригодится. Полнофункциональный вариант бота и необходимые скрипты управления HLDS+HLTV выложу чуть позже, после проведения косметических процедур...
Код:
##################################
# (c) 2023, ACS RedFoxxx Edition #
##################################
def acs_rcon_cmd(addr, acs_global_password, cmd):
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
message = b'\xff\xff\xff\xffchallenge rcon'
s.settimeout(1);
s.sendto(message, addr)
try:
data, acs_global_address = s.recvfrom(1024)
except Exception as err:
return acs_err_no_response_from % addr[0]+':'+str(addr[1])
if data[0:13] == b'\xff\xff\xff\xffchallenge':
challenge = data[19:-2]
message = b'\xff\xff\xff\xffrcon ' + challenge + b' ' + acs_global_password.encode() + b' ' + cmd.encode() + b'\n'
s.sendto(message, addr)
buffer = b''
while True:
try:
data, acs_global_address = s.recvfrom(4096)
except Exception:
break;
if len(data) > 0:
if data[0:4] == b'\xfe\xff\xff\xff':
data = data[9:]
if data[0:4] == b'\xff\xff\xff\xff':
data = data[4:]
if data[0] == 0x6c:
data = data[1:-2]
buffer += data
else:
break
return buffer.decode().rstrip('\n')
else:
return acs_err_challenge_not_aviable % data
Код:
#!/usr/bin/python3
##################################
# (c) 2023, ACS RedFoxxx Edition #
##################################
import socket, MySQLdb, MySQLdb.cursors, re
import asyncio
from ctypes import *
from pprint import pprint
def acs_build_arr(arr, key_fields, row, is_loop=False):
tmp_arr={}
_k1 = _k2 = -1
if type(arr) is list:
if len(arr) == 0:
arr.append({})
arr = arr[-1]
else:
_t = arr[-1]
_k1 = _t[list(_t)[0]]
_k2 = row[list(_t)[0]]
if _k1 != _k2:
arr.append({})
arr = arr[-1]
for i, val in key_fields.items():
if val != 0:
if i not in arr or is_loop:
if type(arr) is list:
arr = arr[-1][i]
_t = arr[-1]
_k3 = _t[list(_t)[0]]
_k4 = row[list(_t)[0]]
if _k3 != _k4:
arr.append({})
arr = arr[-1]
tmp_arr = acs_build_arr(arr, val, row, True)
continue
else:
arr[i] = []
tmp_arr = acs_build_arr(arr[i], val, row, True)
else:
tmp_arr[i] = row[i]
if i not in arr:
if _k1 != _k2 and (_k1 > 0 and _k2 > 0) or (_k1 < 0 and _k2 < 0):
arr[i] = row[i]
return tmp_arr
localhost = socket.gethostbyname(socket.gethostname())
print('Local IP is %s' % localhost)
mysql_host = ""
mysql_db = ""
mysql_user = ""
mysql_pass = ""
acs_tg_user_index = "telegram"
acs_global_servers = list()
acs_global_services = {}
acs_global_users = {}
def acs_check_database_flags(id, flags="", is_strict_check=False):
result = is_strict_check
if flags:
if id in acs_global_users:
for adm in acs_global_users[id]["admins"]:
for svc in adm["services"]:
s_flag = svc["s_rights"].replace("_nick_prefix", "")
if s_flag:
for f in list(flags):
if is_strict_check:
if f not in s_flag:
result = False
break
else:
result |= (f in s_flag)
if result:
break
else:
result = False
else:
result = False
return result
def acs_copy_dict_list(lst, data):
for row in data:
lst.append({})
_arr=lst[-1]
for key in row:
_arr[key] = row[key]
def acs_read_database(servers, services, users):
print("Connection to the Database...")
try:
db = MySQLdb.connect(host = mysql_host, # your host, usually localhost
user = mysql_user, # your username
passwd = mysql_pass, # your password
db = mysql_db, # name of the data base
charset = "utf8",
init_command = "SET NAMES UTF8",
cursorclass = MySQLdb.cursors.DictCursor)
except Exception as err:
print("Connected error with %s" % err)
return False
cur = db.cursor()
cur.execute("select name, ip, port, rcon_password from servers where rcon_password != ''")
acs_copy_dict_list(servers, cur.fetchall())
if (not servers):
print('Error! Servers not found, check MySQL parametrs')
return False
print('Finded %d server(s)' % len(servers))
cur.execute("select name, rights from services")
rows=cur.fetchall()
if not rows:
print('Error! Services not found, check MySQL parametrs')
#exit(-1)
for row in rows:
acs_global_services[row['rights']] = row['name']
#pprint(acs_global_services)
#exit(0)
#acs_copy_dict_list(services, cur.fetchall())
if (not services):
print('Error! Services not found, check MySQL parametrs')
return False
print("Found %s service(s)" % len(acs_global_services))
#pprint(acs_global_services)
sql="select u.id as id, u.name, u.nick, u.email, u.vk, u.telegram, u.steam_id, u.prefix, a.id as a_id, a.name as a_name, a.pass as a_pass, a.type as a_type, a.server as a_server, a.active as a_active, a_svc.id as s_id, a_svc.service as s_service, a_svc.service_time as s_service_time, a_svc.bought_date as s_bought_date, a_svc.ending_date as s_ending_time, svc.name as s_name, svc.rights as s_rights, svc.server as s_server from users as u, admins as a, admins__services as a_svc, services as svc where a.user_id = u.id and a_svc.admin_id = a.id and a_svc.service = svc.id and u.telegram <> '' order by u.id, a.id, svc.id"
#print("SQL = %s" % sql )
cur.execute(sql)
user_key_fields = {
"id" : 0, "name" : 0, "nick" : 0, "email" : 0, "vk" : 0, "telegram" : 0, "steam_id" : 0, "prefix" : 0,
"admins" : {
"a_id" : 0, "a_name" : 0, "a_pass" : 0, "a_type" : 0, "a_server" : 0, "a_active" : 0,
"services" : {
"s_id" : 0, "s_service" : 0, "s_service_time" : 0, "s_bought_date" : 0, "s_ending_time" : 0, "s_name" : 0, "s_rights" : 0, "s_server" : 0
}
}
}
rows = cur.fetchall()
if (not rows):
print("Error! Users with TG not found, check MySQL parametrs")
return False
print("Found %s user(s)" % len(rows))
for row in rows:
_index=row[acs_tg_user_index]
if _index == '' or not _index.isnumeric():
continue
if _index not in users:
users[_index] = {}
acs_build_arr(users[_index], user_key_fields, row)
db.close()
return True
Последнее редактирование: