How to Programmatically Retrieve Active OpenVPN User Sessions in Django on Ubuntu


2 views

OpenVPN provides a built-in management interface that outputs real-time connection data. For developers building custom VPN management dashboards, this interface is the most reliable way to extract active session information.

First, edit your OpenVPN server configuration (typically in /etc/openvpn/server.conf) to add these lines:

management 127.0.0.1 7505
management-client-auth
management-query-proxy

Restart OpenVPN for changes to take effect:

sudo service openvpn restart

Here's a Django view function that connects to the management interface and parses active connections:

import socket
from django.http import JsonResponse

def get_vpn_sessions(request):
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.connect(('127.0.0.1', 7505))
        s.send(b'status 2\n')
        
        data = b''
        while True:
            chunk = s.recv(4096)
            if not chunk:
                break
            data += chunk
        s.close()
        
        # Parse the status output
        clients = []
        for line in data.decode().split('\n'):
            if line.startswith('CLIENT_LIST'):
                parts = line.split(',')
                clients.append({
                    'common_name': parts[1],
                    'real_address': parts[2],
                    'bytes_received': parts[3],
                    'bytes_sent': parts[4],
                    'connected_since': parts[5]
                })
        
        return JsonResponse({'status': 'success', 'clients': clients})
    
    except Exception as e:
        return JsonResponse({'status': 'error', 'message': str(e)})

For systems where enabling the management interface isn't possible, you can parse OpenVPN's log files:

import re
from collections import defaultdict

def parse_openvpn_log(log_path='/var/log/openvpn.log'):
    active_clients = defaultdict(dict)
    
    with open(log_path) as f:
        for line in f:
            # Match connection events
            connect_match = re.search(
                r'(.+?) .+? (.+?) $$(.+?)$$ Peer Connection Initiated', 
                line
            )
            if connect_match:
                timestamp, client_ip, common_name = connect_match.groups()
                active_clients[common_name] = {
                    'ip': client_ip,
                    'connect_time': timestamp.strip()
                }
            
            # Match disconnection events
            disconnect_match = re.search(
                r'(.+?) .+? $$(.+?)$$ $$.+$$ Inactivity timeout', 
                line
            )
            if disconnect_match:
                timestamp, common_name = disconnect_match.groups()
                if common_name in active_clients:
                    del active_clients[common_name]
    
    return active_clients

When implementing this in production:

  • Cache the results for 10-30 seconds to avoid hitting the management interface too frequently
  • Use Django's cache framework to store parsed results
  • Consider implementing a background worker if parsing large log files
  • For high-traffic VPN servers, consider writing the output to a temporary file and reading that instead

Remember to:

  • Restrict access to the management interface (iptables/nftables)
  • Never expose the management interface to external networks
  • Use proper authentication for your Django views
  • Sanitize all output before displaying in web views

When building a Django-based OpenVPN management interface, retrieving active connection data requires interacting with OpenVPN's management interface. The most reliable method is through the OpenVPN management console, which provides real-time connection data.

# First enable management interface in OpenVPN config
management 127.0.0.1 7505

Use netcat or similar tools to connect to the OpenVPN management port:

echo "status 2" | nc 127.0.0.1 7505

This returns a structured output containing:

  • Connected client IPs
  • Connection timestamps
  • Bytes transferred
  • Client certificates

Here's a Django utility function to parse active connections:

import socket
from collections import defaultdict

def get_openvpn_connections(management_ip='127.0.0.1', port=7505):
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.connect((management_ip, port))
        s.send(b"status 2\n")
        data = s.recv(4096).decode('utf-8')
        s.close()
        
        connections = []
        for line in data.split('\n'):
            if line.startswith('CLIENT_LIST'):
                parts = line.split(',')
                connections.append({
                    'common_name': parts[1],
                    'real_address': parts[2],
                    'bytes_received': parts[3],
                    'bytes_sent': parts[4],
                    'connected_since': parts[5]
                })
        return connections
    except Exception as e:
        print(f"Error fetching OpenVPN status: {e}")
        return []

For systems without management interface access, parse OpenVPN's log files:

import re

def parse_openvpn_log(log_path='/var/log/openvpn.log'):
    pattern = re.compile(r'(.+?) (.+?) (.+?) (.+?) (.+?) MULTI: (.+?)')
    connections = []
    
    with open(log_path) as f:
        for line in f:
            if "MULTI: primary virtual IP" in line:
                match = pattern.search(line)
                if match:
                    connections.append({
                        'timestamp': match.group(1),
                        'client_ip': match.group(6).split()[0]
                    })
    return connections

Store connection data in Django models:

from django.db import models

class VPNConnection(models.Model):
    common_name = models.CharField(max_length=255)
    client_ip = models.GenericIPAddressField()
    bytes_received = models.BigIntegerField()
    bytes_sent = models.BigIntegerField()
    connection_start = models.DateTimeField()
    last_updated = models.DateTimeField(auto_now=True)
    
    class Meta:
        indexes = [
            models.Index(fields=['common_name']),
            models.Index(fields=['client_ip']),
        ]

For a responsive UI, consider using Django Channels:

# consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer

class VPNMonitorConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await self.accept()
        while True:
            connections = get_openvpn_connections()
            await self.send(json.dumps(connections))
            await asyncio.sleep(5)
  • Always restrict management interface access
  • Use TLS for management interface communication
  • Implement proper authentication
  • Regularly rotate management interface credentials