Zdalne-systemy-z-kit-uVPN/utils.py

101 lines
2.3 KiB
Python
Raw Normal View History

2023-04-14 10:37:37 +00:00
import datetime
2023-04-13 09:52:51 +00:00
import hashlib
import secrets
2023-04-13 11:12:08 +00:00
import string
import random
2023-04-14 08:41:07 +00:00
import os
2023-04-14 10:37:37 +00:00
import subprocess
import threading
from time import sleep
import db
2023-04-14 12:02:01 +00:00
import config
import ipaddress
2023-04-25 08:24:30 +00:00
from os import chmod
from Crypto.PublicKey import RSA
2023-04-19 09:45:08 +00:00
2023-04-25 10:32:05 +00:00
DELETE_TIMEOUT = 45
RESTART_DELETE_THREAD = 3
2023-04-21 14:39:25 +00:00
2023-04-13 11:12:08 +00:00
def generate_random_string(length):
2023-04-13 11:23:54 +00:00
letters = string.ascii_letters
2023-04-13 11:12:08 +00:00
random_string = ''.join(random.choice(letters) for i in range(length))
return random_string
2023-04-13 09:52:51 +00:00
2023-04-19 09:45:08 +00:00
2023-04-13 09:52:51 +00:00
def hash_password(password):
return hashlib.sha512(password.encode('utf-8')).hexdigest()
2023-04-19 09:45:08 +00:00
2023-04-13 09:52:51 +00:00
def generate_auth_token():
2023-04-14 08:41:07 +00:00
return secrets.token_urlsafe(32)
2023-04-19 09:45:08 +00:00
2023-04-14 08:41:07 +00:00
def ping_client(ip):
2023-04-14 09:57:27 +00:00
response = os.system("ping -c 1 " + ip + "> /dev/null")
2023-04-14 08:41:07 +00:00
if response == 0:
return True
else:
2023-04-14 10:37:37 +00:00
return False
def ssh_thread_function():
2023-04-27 08:17:28 +00:00
subprocess.run(['wssh', '--fbidhttp=False', '--port='+config.webssh_port])
2023-04-19 09:45:08 +00:00
2023-04-14 10:37:37 +00:00
def check_allocation_thread_function():
while True:
ids = db.get_image_allocation_all()
2023-04-21 14:39:25 +00:00
if ids is not None:
for x in ids:
ip = db.get_image_allocation_clientip_id_vpn(x[0])
ping_thread = PingThread(ip, x[0])
ping_thread.start()
2023-04-19 09:45:08 +00:00
sleep(RESTART_DELETE_THREAD)
2023-04-19 09:45:08 +00:00
2023-04-14 10:37:37 +00:00
class PingThread(threading.Thread):
def __init__(self, ip, id):
super(PingThread, self).__init__()
self.Ip = ip
self.Id = id
2023-04-19 09:45:08 +00:00
2023-04-14 10:37:37 +00:00
def run(self):
if self.Ip is None:
return
if ping_client(self.Ip) == False:
date = db.get_image_allocation_time_id(self.Id)
if date is None:
return
2023-04-21 08:59:34 +00:00
delta = datetime.datetime.utcnow() - date
if delta.total_seconds() > DELETE_TIMEOUT:
2023-04-14 10:37:37 +00:00
db.del_image_allocation_id(self.Id)
2023-04-25 10:27:26 +00:00
print("deleted:", self.Id)
2023-04-14 10:37:37 +00:00
else:
db.update_image_allocation_time(self.Id)
2023-04-19 09:45:08 +00:00
def init_threads():
2023-04-14 10:37:37 +00:00
ssh_thread = threading.Thread(target=ssh_thread_function)
ssh_thread.start()
2023-04-19 09:45:08 +00:00
allocation_thread = threading.Thread(
target=check_allocation_thread_function)
allocation_thread.start()
2023-04-21 14:39:25 +00:00
def is_valid_ip_address(ip: str) -> bool:
try:
ipaddress.IPv4Address(ip)
return True
except ipaddress.AddressValueError:
2023-04-24 06:27:35 +00:00
try:
ipaddress.IPv6Address(ip)
return True
except:
pass
return False