import socket import sys import re from ipaddress import IPv4Address from datetime import datetime import mysql.connector import json import requests import pandas as pd try: import geoip2.database from geoip2.errors import AddressNotFoundError geoip2_available = True except ImportError: geoip2_available = False class OpenVPNMonitor: def __init__(self, server_ip, server_port, data_limit_mb, db_conn, log_file='disconnect_log.csv'): self.server_ip = server_ip self.server_port = server_port self.data_limit_bytes = data_limit_mb * 1024 * 1024 self.s = None self.log_file = log_file self.db_conn = db_conn self.connected_servers = [] self.gi = geoip2.database.Reader('/var/www/html/ovpn-moniter/GeoLite2-City.mmdb') self.geoip_version = 2 def _socket_recv(self, length): if sys.version_info[0] == 2: return self.s.recv(length) else: return self.s.recv(length).decode('utf-8') def connect(self): try: self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.s.connect((self.server_ip, self.server_port)) print(f"Connected to OpenVPN management interface at {self.server_ip}:{self.server_port}") except Exception as e: print(f"Error connecting to OpenVPN management: {e}") def get_status(self): try: if not self.s: print("Socket is not connected!") return None self.s.sendall(b"status\n") data = '' while True: socket_data = self._socket_recv(4096) # print(socket_data) if not socket_data: break socket_data = re.sub(r'>INFO(.)*\r\n', '', socket_data) data += socket_data if 'END' in socket_data: break if not data: print("Received empty data from server.") return None return data except Exception as e: print(f"Error receiving data: {e}") return None 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 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 get_geoip_data(self,session, gi, geoip_version): try: if geoip_version == 1: # Legacy GeoIP lookup gir = 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 = 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: # Handle the case where the IP address is not found in the database pass except SystemError: # Handle any system errors that might occur pass def parse_status(self, data, 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() result = { 'host': self.server_ip, 'name': 'UDP', 'port': str(self.server_port), 'socket_connected': True, 'release': 'OpenVPN 2.6.11 x86_64-alpine-linux-musl [SSL (OpenSSL)] [LZO] [LZ4] [EPOLL] [MH/PKTINFO] [AEAD]', 'version': '2.6.11', 'sessions': {} } parsing_clients = False parsing_routing_table = False for line in lines: # print('---------------linelineline',line) # 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 if parsing_clients and line: # CLIENT_LIST section 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 # Initialize session dictionary with basic fields session = { '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': '' } self.get_geoip_data(session, self.gi, self.geoip_version) # Store the session in the result dict result['sessions'][remote_ip] = session elif parsing_routing_table and line: # ROUTING TABLE section if line.startswith('Virtual Address'): continue columns = line.split(',') if len(columns) >= 4: local_ip = columns[0].strip() last_seen = columns[3].strip() # Assuming you want to add this data to the session dictionary session['local_ip'] = local_ip session['last_seen'] = last_seen # After processing ROUTING TABLE, we can stop parsing it parsing_routing_table = False return result def n_sessions(self, vpn): """Update session data into the database.""" try: with self.db_conn.cursor() as db_cursor: sessions = vpn['sessions'] connected_servers_str = "" for local_ip, session_data in sessions.items(): username = session_data['username'] remote_ip = str(session_data['remote_ip']) incoming_bytes_recv = session_data['bytes_recv'] incoming_bytes_sent = session_data['bytes_sent'] connected_since = session_data['connected_since'] total_data_usage = session_data['total_data_usage'] last_seen = 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'] incoming_total_bytes = incoming_bytes_recv + incoming_bytes_sent db_cursor.execute("SELECT bytes_recv, bytes_sent, total_bytes FROM ovpn_server_logs WHERE username = %s", (username,)) result = db_cursor.fetchone() # print('result',result) if result: self.connected_servers.append(username) connected_servers_str = "'" + "', '".join(self.connected_servers) + "'" current_bytes_recv, current_bytes_sent, current_total_bytes = result new_total_bytes = incoming_bytes_recv + incoming_bytes_sent if (current_total_bytes+new_total_bytes) > self.data_limit_bytes: print(f"User {username} has exceeded the data limit ({current_total_bytes+new_total_bytes} bytes). Disconnecting...") disconnect_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') self.log_disconnect(username, remote_ip, current_total_bytes+new_total_bytes, disconnect_time) self.disconnect_user(username, remote_ip, current_total_bytes+new_total_bytes) update_query = f"UPDATE ovpn_server_logs SET is_latest = 0 WHERE username = '{username}'" db_cursor.execute(update_query) self.db_conn.commit() else: if current_total_bytes < new_total_bytes: print('--------------') update_query = """ UPDATE ovpn_server_logs SET bytes_recv = %s, bytes_sent = %s, 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 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_ip, self.server_ip, username)) self.db_conn.commit() print(f"Updated live data usage BEFORE for {username}: recv = {current_bytes_recv}, sent = {current_bytes_sent}, total = {current_total_bytes}") print(f"Updated live data usage AFTER for {username}: recv = {incoming_bytes_recv}, sent = {incoming_bytes_sent}, total = {new_total_bytes}") else: update_query = """ UPDATE ovpn_server_logs SET bytes_recv = %s, bytes_sent = %s, 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 WHERE username = %s""" db_cursor.execute(update_query, (incoming_bytes_recv+current_bytes_recv, incoming_bytes_sent + current_bytes_sent, new_total_bytes+current_total_bytes, 1, connected_since, last_seen, region, latitude, city, country, longitude, remote_ip, local_ip, self.server_ip, username)) self.db_conn.commit() print(f"Updated live data usage BEFORE for {username}: recvv = {current_bytes_recv}, sent = {current_bytes_sent}, total = {current_total_bytes}") print(f"Updated cumulative data usage AFTER for {username}: recvv = {incoming_bytes_recv}, sent = {incoming_bytes_sent}, total = {new_total_bytes}") else: try: insert_query = """ INSERT INTO ovpn_server_logs (username, bytes_recv, bytes_sent, total_bytes ,is_latest ,connected_since , last_seen, region, latitude, city, country, longitude, remote_ip, local_ip, server_ip) VALUES (%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_total_bytes, 1, connected_since, last_seen, region, latitude, city, country, longitude, remote_ip, local_ip, self.server_ip)) 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: print(e) if not connected_servers_str: connected_servers_str = "''" db_cursor.execute(f"""SELECT * FROM ovpn_server_logs WHERE username NOT IN ({connected_servers_str}) AND is_latest = 1""") connected_servers_data = db_cursor.fetchall() if connected_servers_data: df = pd.DataFrame(connected_servers_data, columns=['username','bytes_recv','bytes_sent','total_bytes','is_latest','remote_ip','country','region','city','local_ip','latitude','longitude','connected_since','last_seen','server_ip']) user_data_dict = df.set_index('username').T.to_dict() print('user_data_dict',user_data_dict) if connected_servers_data: batch_size = 100 total_records = len(connected_servers_data) for start in range(0, total_records, batch_size): end = min(start + batch_size, total_records) batch_users = connected_servers_data[start:end] usernames_str = ",".join(f"'{user[0]}'" for user in batch_users) update_query = f"UPDATE ovpn_server_logs SET is_latest = 0 WHERE username IN ({usernames_str})" db_cursor.execute(update_query) self.db_conn.commit() print(f"Updated is_latest = 0 for batch: {usernames_str}") else: pass except mysql.connector.Error as err: print(f"Error: {err}") self.db_conn.rollback() def fetch_data_once(self): data = self.get_status() if data: # Parse the raw data into structured form structured_data = self.parse_status(data,self.gi) if structured_data and 'sessions' in structured_data: # Update the database with the session data self.n_sessions(structured_data) return structured_data else: print("No data to process.") return None if __name__ == "__main__": db_conn = mysql.connector.connect( host="13.200.64.163", user="root", password="eternal123", database="ovpn_moniter", port=5306 ) vpn_monitor = OpenVPNMonitor("13.200.64.163", 5555, data_limit_mb=2000, db_conn=db_conn) vpn_monitor.connect() structured_data = vpn_monitor.fetch_data_once() if structured_data: print(structured_data)