CVE-2021-31606: Improper Authentication (CWE-287)
value: furlongm openvpn-monitor through 1.1.3 allows Authorization Bypass to disconnect arbitrary clients.
CWE description: When an actor claims to have a given identity, the product does not prove or insufficiently proves that the claim is correct.
CVSS v3 base score: 7.5
Published: 2021-09-27T06:15Z
Repository: https://github.com/furlongm/openvpn-monitor
Language: Python
Files changed: openvpn-monitor.py
Commit date: 2021-07-01 15:44:59 -0400
Commit message: only allow disconnect if feature is enabled and sanitize args (#204)


Diff: diff --git a/openvpn-monitor.py b/openvpn-monitor.py
index bf0a83a..83f603a 100755
--- a/openvpn-monitor.py
+++ b/openvpn-monitor.py
@@ -178,22 +178,30 @@ class OpenvpnMgmtInterface(object):
     def __init__(self, cfg, **kwargs):
         self.vpns = cfg.vpns
 
         if kwargs.get('vpn_id'):
             vpn = self.vpns[kwargs['vpn_id']]
-            self._socket_connect(vpn)
-            if vpn['socket_connected']:
-                release = self.send_command('version\n')
-                version = semver(self.parse_version(release).split(' ')[1])
-                if version.major == 2 and \
-                        version.minor >= 4 and \
-                        kwargs.get('client_id'):
-                    command = 'client-kill {0!s}\n'.format(kwargs['client_id'])
-                else:
-                    command = 'kill {0!s}:{1!s}\n'.format(kwargs['ip'], kwargs['port'])
-                self.send_command(command)
-                self._socket_disconnect()
+            disconnection_allowed = vpn['show_disconnect']
+            if disconnection_allowed:
+                self._socket_connect(vpn)
+                if vpn['socket_connected']:
+                    release = self.send_command('version\n')
+                    version = semver(self.parse_version(release).split(' ')[1])
+                    command = False
+                    client_id = int(kwargs.get('client_id'))
+                    if version.major == 2 and \
+                            version.minor >= 4 and \
+                            client_id:
+                        command = 'client-kill {0!s}\n'.format(client_id)
+                    else:
+                        ip = ip_address(kwargs['ip'])
+                        port = int(kwargs['port'])
+                        if ip and port:
+                            command = 'kill {0!s}:{1!s}\n'.format(ip, port)
+                    if command:
+                        self.send_command(command)
+                    self._socket_disconnect()
 
         geoip_data = cfg.settings['geoip_data']
         self.geoip_version = None
         self.gi = None
         try:

Vulnerable code:             self._socket_connect(vpn)
            if vpn['socket_connected']:
                release = self.send_command('version\n')
                version = semver(self.parse_version(release).split(' ')[1])
                if version.major == 2 and \
                        version.minor >= 4 and \
                        kwargs.get('client_id'):
                    command = 'client-kill {0!s}\n'.format(kwargs['client_id'])
                else:
                    command = 'kill {0!s}:{1!s}\n'.format(kwargs['ip'], kwargs['port'])
                self.send_command(command)
                self._socket_disconnect()
Fixed code:             disconnection_allowed = vpn['show_disconnect']
            if disconnection_allowed:
                self._socket_connect(vpn)
                if vpn['socket_connected']:
                    release = self.send_command('version\n')
                    version = semver(self.parse_version(release).split(' ')[1])
                    command = False
                    client_id = int(kwargs.get('client_id'))
                    if version.major == 2 and \
                            version.minor >= 4 and \
                            client_id:
                        command = 'client-kill {0!s}\n'.format(client_id)
                    else:
                        ip = ip_address(kwargs['ip'])
                        port = int(kwargs['port'])
                        if ip and port:
                            command = 'kill {0!s}:{1!s}\n'.format(ip, port)
                    if command:
                        self.send_command(command)
                    self._socket_disconnect()