import asyncio from ipaddress import IPv4Address import mysql.connector import redis import json import requests import time import calendar import os import logging import traceback from datetime import datetime, timedelta try: import geoip2.database from geoip2.errors import AddressNotFoundError geoip2_available = True except ImportError: geoip2_available = False class RedisClientHelper: def __init__(self, host='13.200.64.163', port=6379, db=0, decode_responses=True): """Initialize the Redis client.""" self.redis_client = redis.StrictRedis(host=host, port=port, db=db, decode_responses=decode_responses) self.keys = [] self.disconnected_data = [] def scan_all_keys(self, pattern='*', count=1000): """Scan all keys in Redis.""" cursor = 0 while True: cursor, new_keys = self.redis_client.scan(cursor=cursor, match=pattern, count=count) self.keys.extend(new_keys) print('keys getting from redis',self.keys) if cursor == 0: break return self.keys def get_all_data(self, keys): """Get all data from Redis.""" all_data = {} for key in keys: key_type = self.redis_client.type(key) print('-key_type',key_type) if key_type == 'hash': data = self.redis_client.hgetall(key) all_data[key] = data print('======a===',type(data['is_deleted'])) if 'is_deleted' in data and int(data['is_deleted']) == 0: self.disconnected_data.append(key) elif key_type == 'string': print('======b==') all_data[key] = self.redis_client.get(key) elif key_type == 'list': all_data[key] = self.redis_client.lrange(key, 0, -1) elif key_type == 'set': all_data[key] = list(self.redis_client.smembers(key)) elif key_type == 'zset': all_data[key] = self.redis_client.zrange(key, 0, -1, withscores=True) return all_data class LoggerManager: def __init__(self, log_directory, delete_log_days=10, log_name="VPN"): self.log_directory = log_directory self.delete_log_days = delete_log_days self.log_name = log_name self.logger = None self.setup_logging() self.remove_old_logs() def setup_logging(self): """Sets up the logging configuration and creates the log file.""" # Ensure the log directory exists if not os.path.exists(self.log_directory): os.makedirs(self.log_directory) log_file_path = os.path.join(self.log_directory, f'{self.log_name}_{datetime.now().strftime("%Y-%m-%d")}.log') # Create a log formatter and handler log_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') file_handler = logging.FileHandler(log_file_path) file_handler.setFormatter(log_formatter) # Set up the logger self.logger = logging.getLogger(self.log_name) self.logger.addHandler(file_handler) self.logger.setLevel(logging.INFO) def remove_old_logs(self): """Removes log files older than the specified number of days.""" cutoff_date = datetime.now() - timedelta(days=self.delete_log_days) for filename in os.listdir(self.log_directory): if filename.endswith('.log'): file_path = os.path.join(self.log_directory, filename) file_mod_time = datetime.fromtimestamp(os.path.getmtime(file_path)) # If the file is older than the cutoff date, remove it if file_mod_time < cutoff_date: os.remove(file_path) self.logger.info(f"Removed old log file: {filename}") self.logger.info("Log cleanup complete.") def get_logger(self): """Returns the logger object for use in other classes.""" return self.logger class OpenVPNClient: def __init__(self, servers, db_conn,data_limit_mb,max_concurrent_connections=100): """ Initialize the OpenVPN client with a list of servers, geoip_version, and max concurrent connections. """ self.error_occurred = False self.connected_servers_str = "" self.servers = servers self.gi = geoip2.database.Reader('/var/www/html/1375-staysolid-ovpn-backend/GeoLite2-City.mmdb') self.geoip_version = 2 self.max_concurrent_connections = max_concurrent_connections self.semaphore = asyncio.Semaphore(self.max_concurrent_connections) self.session = {} self.servers_data = [] self.connected_servers = [] self.db_conn = db_conn self.data_limit_bytes = data_limit_mb * 1024 * 1024 self._working_directory = os.path.dirname(os.path.abspath(__file__)) self._log_directory = f'{self._working_directory}/logs' self.logger_manager = LoggerManager(self._log_directory,10,"OVPN_Monitor") self.logger = self.logger_manager.get_logger() # Initialize the Redis client helper self.redis_helper = RedisClientHelper() # Getting data from redis of active server start_time = time.time() all_keys = self.redis_helper.scan_all_keys() self.all_data_retrieved = self.redis_helper.get_all_data(all_keys) print('redis data',self.all_data_retrieved) end_time = time.time() print(f"Time taken to scan {len(all_keys)} keys and retrieve data: {end_time - start_time:.4f} seconds") def disconnect_user(self, username, remote_ip, total_data_usage): try: disconnect_command = f"kill {username}\n" self.s.sendall(disconnect_command.encode('utf-8')) print(f"Sent disconnect command for user: {username}") except Exception as e: print(f"Error sending disconnect command: {e}") def log_disconnect(self, username, remote_ip, total_data_usage, disconnect_time): log_entry = f"{disconnect_time},{username},{remote_ip},{total_data_usage}\n" with open(self.log_file, 'a') as log_file: log_file.write(log_entry) print(f"Logged disconnect: {log_entry.strip()}") def n_sessions(self): # self.logger.info('----------------2') """Update session data into the database.""" try: with self.db_conn.cursor(dictionary=True) as db_cursor: # sessions = vpn['sessions'] disconnected_users = [] self.logger.info(f'Total servers : {len(self.servers_data)}') self.logger.info(self.servers_data) for session_data in self.servers_data: username = session_data['username'] print(']]]]]]]]]]]]]]]]]]]]]]]]]]',self.redis_helper.disconnected_data) if username in self.redis_helper.disconnected_data: self.redis_helper.redis_client.hset(username, "is_deleted", 1) if username != 'UNDEF': remote_ip = str(session_data['remote_ip']) incoming_bytes_recv = session_data['bytes_recv'] incoming_bytes_sent = session_data['bytes_sent'] connected_since = str(session_data['connected_since']) total_data_usage = session_data['total_data_usage'] last_seen = str(session_data['last_seen']) location = session_data['location'] region = session_data['region'] latitude = session_data['latitude'] city = session_data['city'] country = session_data['country'] longitude = session_data['longitude'] local_ips = session_data['local_ip'] server_ip = session_data['host_name'] db_cursor.execute("SELECT o.username, os.name, o.total_bytes_recv, o.total_bytes_sent, o.total_bytes, o.current_bytes_recv, o.current_bytes_sent, o.current_total_bytes ,os.user_device_id, o.user_device_id AS udid FROM ovpn_server_logs AS o JOIN ovpn_servers AS os ON o.username = os.name WHERE o.username = %s", (username,)) result = db_cursor.fetchone() if result: print('if result for ',result['username']) file_name, user_device_id = result['username'],result['user_device_id'] # self.logger.info('insert2') self.logger.info(f"Connected file : {file_name} and User device ID : {user_device_id}") self.connected_servers.append(username) self.connected_servers_str = "'" + "', '".join(self.connected_servers) + "'" new_total_bytes = incoming_bytes_recv + incoming_bytes_sent if result['current_total_bytes'] > new_total_bytes and username not in self.redis_helper.disconnected_data: print('working after same connection at a time1') if username == result['name'] and result['user_device_id'] == result['udid']: print('working after same connection at a time 2') update_query = """UPDATE ovpn_server_logs SET total_bytes_recv = %s, total_bytes_sent = %s, total_bytes = %s WHERE username = %s""" db_cursor.execute(update_query, (result['current_bytes_recv']+result['total_bytes_recv'],result['current_bytes_sent']+result['total_bytes_sent'],result['current_total_bytes']+result['total_bytes'],username)) self.db_conn.commit() print('-----1------') dt = time.gmtime() ts = calendar.timegm(dt) update_query_base = """ UPDATE ovpn_server_logs SET current_bytes_recv = %s, current_bytes_sent = %s, current_total_bytes = %s, is_latest = %s, connected_since = %s, last_seen = %s, region = %s, latitude = %s, city = %s, country = %s, longitude = %s, remote_ip = %s, local_ip = %s, server_ip = %s, updated_at = %s """ # Add user_device_id to the query if it's not None if user_device_id is not None: update_query_base += ", user_device_id = %s" # Add the WHERE clause update_query_base += " WHERE username = %s" # Define the parameters for the query params = [incoming_bytes_recv, incoming_bytes_sent, new_total_bytes, 1, connected_since, last_seen, region, latitude, city, country, longitude, remote_ip, local_ips, server_ip, ts] # Add user_device_id to the parameters if it's not None if user_device_id is not None: params.append(user_device_id) # Append the username to the parameters params.append(username) # self.logger.info(f'==={update_query_base}----{params}') # Execute the query with the adjusted parameters db_cursor.execute(update_query_base, tuple(params)) print('update_query:', update_query_base) self.db_conn.commit() else: try: db_cursor.execute("SELECT os.user_device_id,os.name FROM ovpn_servers AS os WHERE os.name = %s", (username,)) user_data = db_cursor.fetchone() if user_data == None: new_total_bytes = incoming_bytes_recv + incoming_bytes_sent self.logger.info(f'This file is not in OVPN SERVER Please check : {username}') dt = time.gmtime() ts = calendar.timegm(dt) self.connected_servers.append(username) self.connected_servers_str = "'" + "', '".join(self.connected_servers) + "'" update_query = """ UPDATE ovpn_server_logs SET current_bytes_recv = %s, current_bytes_sent = %s, current_total_bytes = %s, is_latest = %s, connected_since = %s, last_seen = %s, region = %s, latitude = %s, city = %s, country = %s, longitude = %s, remote_ip = %s, local_ip = %s, server_ip = %s, updated_at = %s, is_latest = 1 WHERE username = %s""" db_cursor.execute(update_query, (incoming_bytes_recv, incoming_bytes_sent, new_total_bytes, 1, connected_since, last_seen, region, latitude, city, country, longitude, remote_ip, local_ips, server_ip,ts, username)) print('update_query : ',update_query) self.db_conn.commit() else: # self.logger.info(f'New server coming : {user_data}') print('else========') user_d_id = user_data['user_device_id'] self.logger.info(f"New Connected file : {user_data['name']} and User device ID : {user_data['user_device_id']}") # self.logger.info('insert') self.connected_servers.append(username) self.connected_servers_str = "'" + "', '".join(self.connected_servers) + "'" dt = time.gmtime() ts = calendar.timegm(dt) insert_query = """ INSERT INTO ovpn_server_logs (username, current_bytes_recv, current_bytes_sent, current_total_bytes,is_latest ,connected_since , last_seen, region, latitude, city, country, longitude, remote_ip, local_ip, server_ip, created_at, updated_at, user_device_id) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ db_cursor.execute(insert_query, (username, incoming_bytes_recv, incoming_bytes_sent, incoming_bytes_recv + incoming_bytes_sent, 1, connected_since, last_seen, region, latitude, city, country, longitude, remote_ip, local_ips, server_ip, ts, ts, user_d_id)) self.db_conn.commit() print(f"Inserted new data usage record for {username}: recv = {incoming_bytes_recv}, sent = {incoming_bytes_sent}") except Exception as e: self.logger.error('error',e) print(e) else: self.logger.error(f"USER FILE UNDF : {str(session_data['remote_ip'])}") self.connected_servers_str_redis = "'" + "', '".join(self.redis_helper.disconnected_data) + "'" print('keys after add into string',self.connected_servers_str_redis) if not self.connected_servers_str: print('---------IN') self.connected_servers_str = "''" print('These are connected servers',self.connected_servers_str) print("server into redis",self.connected_servers_str_redis) db_cursor.execute(f"""SELECT username, current_bytes_recv, current_bytes_sent, current_total_bytes, total_bytes_recv, total_bytes_sent, total_bytes FROM ovpn_server_logs WHERE username NOT IN ({self.connected_servers_str}) AND username NOT IN ({self.connected_servers_str_redis}) AND is_latest = 1""") connected_servers_data = db_cursor.fetchall() print('Server for disconnect',connected_servers_data) if connected_servers_data: update_data = [] for row in connected_servers_data: username = row['username'] redis_exists = self.redis_helper.redis_client.exists(username) print(redis_exists) if redis_exists: print('into redis condition') redis_data = self.all_data_retrieved.get(username, {}) # redis_data = json.loads(redis_data) update_query = """UPDATE ovpn_server_logs SET current_bytes_recv = %s, current_bytes_sent = %s, current_total_bytes = %s WHERE username = %s""" db_cursor.execute(update_query, (row['current_bytes_recv'] + int(redis_data.get('current_bytes_recv', 0)),row['current_bytes_sent'] + int(redis_data.get('current_bytes_sent', 0)),row['current_total_bytes']) + int(redis_data.get('current_total_bytes', 0)),username) self.db_conn.commit() total_bytes_recv = row['total_bytes_recv'] + row['current_bytes_recv'] + int(redis_data.get('current_bytes_recv', 0)) total_bytes_sent = row['total_bytes_sent'] + row['current_bytes_sent'] + int(redis_data.get('current_bytes_sent', 0)) total_bytes = row['total_bytes'] + row['current_total_bytes'] + int(redis_data.get('current_total_bytes', 0)) update_data.append((total_bytes_recv, total_bytes_sent, total_bytes, username)) self.redis_helper.redis_client.delete(username) else: print('normal condition') total_bytes_recv = row['total_bytes_recv'] + row['current_bytes_recv'] total_bytes_sent = row['total_bytes_sent'] + row['current_bytes_sent'] total_bytes = row['total_bytes'] + row['current_total_bytes'] # Add the updated data to the list update_data.append((total_bytes_recv, total_bytes_sent, total_bytes, username)) if update_data: update_query = """UPDATE ovpn_server_logs SET total_bytes_recv = %s, total_bytes_sent = %s, total_bytes = %s, current_bytes_recv = 0, current_bytes_sent = 0, current_total_bytes = 0, is_latest = 0 WHERE username = %s""" db_cursor.executemany(update_query, update_data) self.db_conn.commit() except mysql.connector.Error as err: self.logger.error(err) print(f"Error: {err}") self.db_conn.rollback() def get_geoip_data(self,session, geoip_version): try: if geoip_version == 1: # Legacy GeoIP lookup gir = self.gi.record_by_addr(str(session['remote_ip'])) if gir is not None: session['location'] = gir.get('country_code', '') session['region'] = gir.get('region', '') session['city'] = gir.get('city', '') session['country'] = gir.get('country_name', '') session['longitude'] = gir.get('longitude', '') session['latitude'] = gir.get('latitude', '') elif geoip_version == 2: # GeoIP2 lookup gir = self.gi.city(str(session['remote_ip'])) session['location'] = gir.country.iso_code if gir.country else '' session['region'] = gir.subdivisions.most_specific.iso_code if gir.subdivisions.most_specific else '' session['city'] = gir.city.name if gir.city else '' session['country'] = gir.country.name if gir.country else '' session['longitude'] = gir.location.longitude if gir.location else '' session['latitude'] = gir.location.latitude if gir.location else '' except geoip2.errors.AddressNotFoundError: pass except SystemError: pass def parse_status(self, data, server_ip, server_port, gi): """Parse the raw data and convert it to a structured dictionary.""" if not data or data.strip() == '': print("No valid data to parse.") return {} lines = data.splitlines() parsing_clients = False parsing_routing_table = False for line in lines: # Start parsing the CLIENT_LIST section if line.startswith('Common Name'): parsing_clients = True continue elif line.startswith('ROUTING TABLE'): parsing_clients = False parsing_routing_table = True continue elif line.startswith('GLOBAL STATS'): break if parsing_clients and line: columns = line.split(',') username = columns[0].strip() remote_ip = columns[1].split(':')[0].strip() bytes_received = int(columns[2].strip()) bytes_sent = int(columns[3].strip()) connected_since = datetime.strptime(columns[4].strip(), '%Y-%m-%d %H:%M:%S') total_data_usage = bytes_received + bytes_sent self.session = { 'host_name' : server_ip, 'session_key': f"{username}_{remote_ip}", 'username': username, 'remote_ip': IPv4Address(remote_ip), 'bytes_recv': bytes_received, 'bytes_sent': bytes_sent, 'connected_since': connected_since, 'total_data_usage': total_data_usage, 'last_seen': '', 'location': '', 'region': '', 'city': '', 'country': '', 'longitude': '', 'latitude': '', 'local_ip': '' } self.get_geoip_data(self.session, self.geoip_version) self.servers_data.append(self.session) session_key = f"{username}_{remote_ip}" elif parsing_routing_table and line: if line.startswith('Virtual Address'): continue columns = line.split(',') if len(columns) >= 4: local_ip = columns[0].strip() username = columns[1].strip() remote_ip = columns[2].split(':')[0].strip() last_seen = columns[3].strip() session_key = f"{username}_{remote_ip}" if session_key in self.session['session_key']: self.session['local_ip'] = local_ip self.session['last_seen'] = last_seen async def connect_to_openvpn(self, server_ip, server_port): """ Connect to the OpenVPN management interface, send a command, and parse the response. """ async with self.semaphore: try: reader, writer = await asyncio.open_connection(server_ip, server_port) print(f"Connected to OpenVPN management interface at {server_ip}:{server_port}") command = "status\n" writer.write(command.encode()) await writer.drain() response = "" while True: chunk = await reader.read(4096) if not chunk: break response += chunk.decode('utf-8') if "END" in chunk.decode('utf-8'): break if not response: print(f"No data received from {server_ip}:{server_port}") return print('response from socket',response) self.parse_status(response, server_ip, server_port, gi=None) # print(f"Parsed data from {server_ip}:{server_port}:\n{parsed_data}") # self.logger.info('--------------1') writer.close() await writer.wait_closed() except TimeoutError as e: self.logger.error(f"TimeoutError: Failed to connect to {server_ip}:{server_port}. Error: {e}") self.error_occurred = True # Set the flag to True when an error occurs except OSError as e: self.logger.error(f"OSError: Failed to connect to {server_ip}:{server_port}. Error: {e}") self.error_occurred = True # Set the flag to True when an error occurs except Exception as e: self.logger.error(f"Exception: An unexpected error occurred while connecting to {server_ip}:{server_port}. Error: {e}") self.error_occurred = True except Exception as e: excp = traceback.format_exc() print("exp------", excp) self.logger.error(excp) print(f"Failed to connect to {server_ip}:{server_port} - Error: {e}") async def run(self): """ Run the tasks for all servers asynchronously with controlled concurrency. """ tasks = [self.connect_to_openvpn(server_ip, server_port) for server_ip, server_port in self.servers] try: await asyncio.gather(*tasks) except Exception as e: self.logger.error(f"An error occurred while running tasks: {e}") self.error_occurred = True if not self.error_occurred: self.n_sessions() time.sleep(5) # Entry point if __name__ == '__main__': db_conn = mysql.connector.connect( host="13.200.64.163", user="root", password="eternal123", database="ovpnnew", port=5306 ) servers = [ ('13.200.64.163', 5555), ] openvpn_client = OpenVPNClient(servers,max_concurrent_connections=100, data_limit_mb=100000, db_conn=db_conn) asyncio.run(openvpn_client.run()) # # Entry point # if __name__ == '__main__': # db_conn = mysql.connector.connect( # host="ovpn-centrealised-server.cl66yeq0k0b7.us-east-2.rds.amazonaws.com", # user="xeg3ua7ooqu1", # password="ahv3eigee2zo", # database="xeg3ua7ooqu12", # # port=5306 # ) # servers = [ # ('172.245.234.181', 5555), # ] # openvpn_client = OpenVPNClient(servers,max_concurrent_connections=100, data_limit_mb=100000, db_conn=db_conn) # asyncio.run(openvpn_client.run())