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 import pytz try: import geoip2.database from geoip2.errors import AddressNotFoundError geoip2_available = True except ImportError: geoip2_available = False class RedisClientHelper: def __init__(self, host='localhost', port=6379, db=0, password='eternal123', decode_responses=True): """Initialize the Redis client.""" self.redis_client = redis.StrictRedis( host=host, port=port, db=db, password=password, 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.gi = geoip2.database.Reader('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) self.logger.info(f'redis data : {self.all_data_retrieved}') self.logger.info(f'this is updated script for sure') 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 convert_to_timestamp(self,datetime_str): """Convert a date-time string (e.g., '2025-03-20 07:42:26') to a Unix timestamp.""" try: return int(datetime.strptime(datetime_str, "%Y-%m-%d %H:%M:%S").timestamp()) except ValueError as e: print(f"Error: Invalid datetime format - {e}") return None def disconnect_file(self,db_cursor,connected_servers_data): update_data, disconnected_data_analytics, delete_from_still_connected = [], [], [] now_timestamp = int(time.time()) for row in connected_servers_data: username = row['username'] print('Filename : ', username) connected_since_timestamp = int(row['connected_since']) # Data calculation for both conditions (reduces code duplication) current_bytes_recv = row['current_bytes_recv'] current_bytes_sent = row['current_bytes_sent'] current_total_bytes = row['current_total_bytes'] if connected_since_timestamp < now_timestamp: print('into past date') db_cursor.execute("""SELECT current_bytes_recv, current_bytes_sent, current_total_bytes FROM still_connected_files_usage WHERE filename = %s""", (username,)) result = db_cursor.fetchone() # self.logger.info(f"still connceted file : current_bytes_recv {current_bytes_recv} still data : {result['current_total_bytes']}") # Calculate differences only if result exists self.logger.info(f"still connected file : current_bytes_recv {current_bytes_recv if current_bytes_recv is not None else 0} still data : {result.get('current_total_bytes', 0) if result else 0}") still_current_bytes_recv = current_bytes_recv - result['current_bytes_recv'] if result else current_bytes_recv still_current_bytes_sent = current_bytes_sent - result['current_bytes_sent'] if result else current_bytes_sent still_current_total_bytes = current_total_bytes - result['current_total_bytes'] if result else current_total_bytes disconnected_data_analytics.append((username, row['user_device_id'], row['remote_ip'], connected_since_timestamp, row['created_at'], still_current_bytes_recv, still_current_bytes_sent, still_current_total_bytes, now_timestamp)) delete_from_still_connected.append(username) self.logger.info(f"File name : {username} total data usage is : {still_current_total_bytes}") else: disconnected_data_analytics.append((username, row['user_device_id'], row['remote_ip'], connected_since_timestamp, row['created_at'], current_bytes_recv, current_bytes_sent, current_total_bytes, now_timestamp)) self.logger.info(f"File name : {username} total data usage is : {current_total_bytes}") # Redis check and data management redis_data = self.all_data_retrieved.get(username, {}) if self.redis_helper.redis_client.exists(username) else {} # Aggregate data from Redis and row total_bytes_recv = row['total_bytes_recv'] + current_bytes_recv + int(redis_data.get('current_bytes_recv', 0)) total_bytes_sent = row['total_bytes_sent'] + current_bytes_sent + int(redis_data.get('current_bytes_sent', 0)) total_bytes = row['total_bytes'] + current_total_bytes + int(redis_data.get('current_total_bytes', 0)) update_data.append((total_bytes_recv, total_bytes_sent, total_bytes, username)) if redis_data: self.redis_helper.redis_client.delete(username) # Clean up Redis cache # Batch updates for better performance if update_data: db_cursor.executemany("""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 """, update_data) self.logger.info(f"disconncted user data {update_data}") self.db_conn.commit() if disconnected_data_analytics: db_cursor.executemany("""INSERT INTO disconnected_files_usage (filename, user_device_id, remote_ip, connected_since, file_creation_date, current_bytes_recv, current_bytes_sent, current_total_bytes, created_at)VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) """, disconnected_data_analytics) self.db_conn.commit() if delete_from_still_connected: db_cursor.execute( "DELETE FROM still_connected_files_usage WHERE filename IN (%s)" % ", ".join(["%s"] * len(delete_from_still_connected)),delete_from_still_connected) self.db_conn.commit() def n_sessions(self): # self.logger.info('----------------2') """Update session data into the database.""" try: with self.db_conn.cursor(buffered=True,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 = self.convert_to_timestamp(str(session_data['connected_since'])) total_data_usage = session_data['total_data_usage'] last_seen = self.convert_to_timestamp(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, user_device_id, remote_ip, connected_since, created_at, 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() self.logger.info(f'Disconnected servers data : {connected_servers_data}') print('Server for disconnect',connected_servers_data) if connected_servers_data: #calling funtion when user disconnected self.disconnect_file(db_cursor,connected_servers_data) except mysql.connector.Error as err: self.logger.error(err) print(f"Error: {err}") traceback.print_exc() 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())