Merge pull request #6 from tonyseek/feature/tap-support

Add TAP mode support by parsing MAC addresses

Author: shonenada

CHANGES.rst

@@ -1,6 +1,8 @@
 0.2.0 (WIP)
 -----------
 
+- Feature GH-1: Add support to TAP mode of OpenVPN servers by parsing virtual
+  addresses as MAC and IP both.
 - Fix GH-2: **BREAK-COMPATIBILITY** Use real or virtual addresses as the key
   of client_list and routing_table, instead of using common name.
 

openvpn_status/models.py

@@ -1,10 +1,9 @@
 from __future__ import absolute_import
 
 from collections import OrderedDict
-from ipaddress import ip_address
 
 from .descriptors import LabelProperty, name_descriptors
-from .utils import parse_time, parse_peer, parse_filesize
+from .utils import parse_time, parse_peer, parse_vaddr, parse_filesize
 
 
 @name_descriptors
@@ -33,7 +32,7 @@ class Client(object):
 class Routing(object):
     """The OpenVPN routing model."""
 
-    virtual_address = LabelProperty(u'Virtual Address', input_type=ip_address)
+    virtual_address = LabelProperty(u'Virtual Address', input_type=parse_vaddr)
     common_name = LabelProperty(u'Common Name')
     real_address = LabelProperty(u'Real Address', input_type=parse_peer)
     last_ref = LabelProperty(u'Last Ref', input_type=parse_time)

openvpn_status/utils.py

@@ -1,14 +1,18 @@
 from __future__ import absolute_import
 
+import re
 import datetime
 import collections
 import ipaddress
 
 from six import python_2_unicode_compatible as unicode_compatible
 from humanize.filesize import naturalsize
+from netaddr import EUI, mac_unix
 
 
 DATETIME_FORMAT_OPENVPN = u'%a %b %d %H:%M:%S %Y'
+RE_VIRTUAL_ADDR_MAC = re.compile(
+    u'^{0}:{0}:{0}:{0}:{0}:{0}$'.format(u'[a-f0-9]{2}'), re.I)
 
 
 def parse_time(time):
@@ -25,6 +29,12 @@ def parse_peer(peer):
     return PeerAddress(ipaddress.ip_address(host), int(port))
 
 
+def parse_vaddr(virtual_addr):
+    if RE_VIRTUAL_ADDR_MAC.search(virtual_addr):
+        return EUI(virtual_addr, dialect=mac_unix)
+    return ipaddress.ip_address(virtual_addr)
+
+
 def parse_filesize(size):
     if isinstance(size, FileSize):
         return size

setup.py

@@ -40,6 +40,7 @@
     install_requires=[
         'six>=1.9.0',
         'humanize',
+        'netaddr',
     ],
     extras_require={
         ':python_version == "2.7"': [

tests/data/openvpn-status.txt

@@ -5,12 +5,14 @@ [email protected],10.10.10.10:49502,334948,1973012,Thu Jun 18 04:23:03 2015
 [email protected],10.10.10.10:49503,334948,1973012,Thu Jun 18 04:23:03 2015
 [email protected],10.10.10.10:64169,1817262,28981224,Thu Jun 18 04:08:39 2015
 [email protected],10.10.10.10:63414,111183,1202203,Thu Jun 18 07:57:25 2015
+[email protected],10.0.0.100:55712,0,0,Thu Oct 19 20:14:19 2017
 ROUTING TABLE
 Virtual Address,Common Name,Real Address,Last Ref
 192.168.255.118,[email protected],10.10.10.10:63414,Thu Jun 18 08:12:09 2015
 192.168.255.134,[email protected],10.10.10.10:49502,Thu Jun 18 08:12:09 2015
 192.168.255.135,[email protected],10.10.10.10:49503,Thu Jun 18 08:12:09 2015
 192.168.255.126,[email protected],10.10.10.10:64169,Thu Jun 18 08:11:55 2015
+22:1d:63:bf:62:38,[email protected],10.0.0.100:55712,Thu Oct 19 20:14:19 2017
 GLOBAL STATS
 Max bcast/mcast queue length,0
 END

tests/test_parser.py

@@ -1,9 +1,11 @@
 from __future__ import absolute_import
 
 import datetime
+import ipaddress
 
 from pytest import fixture, raises
 from six import text_type
+from netaddr import EUI
 
 from openvpn_status.parser import LogParser, ParsingError
 
@@ -22,8 +24,9 @@ def test_parser(openvpn_status):
     parser = LogParser.fromstring(openvpn_status.read())
     status = parser.parse()
 
-    assert len(status.client_list) == 4
-    assert len(status.routing_table) == 4
+    assert len(status.client_list) == 5
+    assert len(status.routing_table) == 5
+
     assert status.global_stats.max_bcast_mcast_queue_len == 0
     assert status.updated_at == datetime.datetime(2015, 6, 18, 8, 12, 15)
 
@@ -36,11 +39,19 @@ def test_parser(openvpn_status):
     assert client.bytes_received == 334948
     assert client.bytes_sent == 1973012
 
-    routing = status.routing_table[u'192.168.255.134']
-    assert text_type(routing.virtual_address) == u'192.168.255.134'
-    assert routing.common_name == u'[email protected]'
-    assert text_type(routing.real_address) == u'10.10.10.10:49502'
-    assert routing.last_ref == datetime.datetime(2015, 6, 18, 8, 12, 9)
+    tun_routing = status.routing_table[u'192.168.255.134']
+    assert isinstance(tun_routing.virtual_address, ipaddress.IPv4Address)
+    assert text_type(tun_routing.virtual_address) == u'192.168.255.134'
+    assert tun_routing.common_name == u'[email protected]'
+    assert text_type(tun_routing.real_address) == u'10.10.10.10:49502'
+    assert tun_routing.last_ref == datetime.datetime(2015, 6, 18, 8, 12, 9)
+
+    tap_routing = status.routing_table[u'22:1d:63:bf:62:38']
+    assert isinstance(tap_routing.virtual_address, EUI)
+    assert text_type(tap_routing.virtual_address) == u'22:1d:63:bf:62:38'
+    assert tap_routing.common_name == u'[email protected]'
+    assert text_type(tap_routing.real_address) == u'10.0.0.100:55712'
+    assert tap_routing.last_ref == datetime.datetime(2017, 10, 19, 20, 14, 19)
 
 
 def test_parser_with_syntax_errors(broken_status):