Fix #719: convert macaddr in all sorts of forms to std format

Signed-off-by: Dinesh Dutt <[email protected]>
(cherry picked from commit 948f824)

Author: ddutt

suzieq/engines/pandas/address.py

@@ -104,9 +104,8 @@ def get(self, **kwargs) -> pd.DataFrame:
             for i, a in enumerate(addr):
                 if addr_types[i] == 0:
                     # convert the macaddr format to internal format
-                    if '.' in a:
-                        a = convert_macaddr_format_to_colon(a)
-                    macaddr.append(a.lower())
+                    a = convert_macaddr_format_to_colon(a)
+                    macaddr.append(a)
                 elif addr_types[i] == 4:
                     if '/' not in a:
                         a += '/'

suzieq/engines/pandas/network.py

@@ -78,9 +78,7 @@ def _find_address(self, addr: str, **kwargs) -> pd.DataFrame:
 
         # Convert Cisco-style MAC address to standard MAC addr format,
         # and lowercase all letters of the alphabet
-        if '.' in addr:
-            addr = convert_macaddr_format_to_colon(addr)
-        addr = addr.lower()
+        addr = convert_macaddr_format_to_colon(addr)
 
         addr_df = self._find_addr_arp(addr, **kwargs)
         if addr_df.empty:

suzieq/shared/utils.py

@@ -518,8 +518,8 @@ def validate_macaddr(macaddr: str) -> bool:
 
     """
     if isinstance(macaddr, str):
-        if re.fullmatch(r'([0-9a-fA-F]{4}.){2}[0-9a-fA-F]{4}', macaddr) or \
-           re.fullmatch(r'([0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}', macaddr):
+        macaddr = convert_macaddr_format_to_colon(macaddr)
+        if re.fullmatch(r'([0-9a-f]{2}:){5}[0-9a-f]{2}', macaddr):
             return True
 
     return False

suzieq/sqobjects/network.py

@@ -3,7 +3,7 @@
 import pandas as pd
 
 from suzieq.sqobjects.basicobj import SqObject
-from suzieq.shared.utils import (humanize_timestamp,
+from suzieq.shared.utils import (humanize_timestamp, validate_macaddr,
                                  convert_macaddr_format_to_colon)
 
 
@@ -31,9 +31,7 @@ def find(self, **kwargs) -> pd.DataFrame():
                 ip_address(addr)
             except ValueError:
                 addr = convert_macaddr_format_to_colon(addr)
-                if not re.match(
-                        "[0-9a-f]{2}([-:]?)[0-9a-f]{2}(\\1[0-9a-f]{2}){4}$",
-                        addr):
+                if not validate_macaddr(addr):
                     return pd.DataFrame(
                         {'error': [f'Not valid IP or MAC address: {addr}']})
         try:

tests/integration/test_parsing_arpnd.py

@@ -4,6 +4,7 @@
 import pytest
 
 import pandas as pd
+from suzieq.shared.utils import validate_macaddr
 from tests.conftest import validate_host_shape, DATADIR, _get_table_data
 
 
@@ -15,9 +16,7 @@ def validate_arpnd_tbl(df: pd.DataFrame):
     assert (df.state.isin(["permanent", 'reachable', 'router', 'noarp',
                            'failed'])).all()
     pass_df = df.query('state != "failed"')
-    assert pass_df.macaddr.apply(
-        lambda x: re.match("[0-9a-f]{2}([-:]?)[0-9a-f]{2}(\\1[0-9a-f]{2}){4}$",
-                           x) is not None).all()
+    assert pass_df.macaddr.apply(validate_macaddr).all()
     assert not (pass_df.oif.isin(["", "None"])).all()  # noqa
     assert (pass_df.remote.isin([True, False])).all()
 

tests/integration/test_parsing_macs.py

@@ -3,6 +3,7 @@
 import pytest
 import numpy as np
 import pandas as pd
+from suzieq.shared.utils import validate_macaddr
 from tests.conftest import DATADIR, validate_host_shape, _get_table_data
 
 
@@ -13,9 +14,7 @@ def validate_macs(df: pd.DataFrame):
     assert (df.oif != '').all()
     assert (df.mackey != '').all()
     # Validate that the only MAC addresses there are fit the macaddr format
-    assert df.macaddr.apply(
-        lambda x: re.match("[0-9a-f]{2}([-:]?)[0-9a-f]{2}(\\1[0-9a-f]{2}){4}$",
-                           x) is not None).all()
+    assert df.macaddr.apply(validate_macaddr).all()
     # Ignore Linux HER entries and interface MAC entries, and some NXOS entries
     assert (df.query(
         'macaddr != "00:00:00:00:00:00" and flags != "permanent" and '
@@ -46,14 +45,14 @@ def validate_interfaces(df: pd.DataFrame, datadir: str):
     # as routed interfaces that have a VLAN of 0
     only_oifs = df.query('~oif.isin(["bridge", "sup-eth1", '
                          '"vPC Peer-Link", "nve1", "Router"])') \
-        .query('~oif.str.startswith("vtep.")') \
-        .query('vlan != 0') \
-        .groupby(by=['namespace', 'hostname', 'vlan'])['oif'] \
-        .unique() \
-        .reset_index() \
-        .explode('oif') \
-        .rename(columns={'oif': 'ifname'}) \
-        .reset_index(drop=True)
+                  .query('~oif.str.startswith("vtep.")') \
+                  .query('vlan != 0') \
+                  .groupby(by=['namespace', 'hostname', 'vlan'])['oif'] \
+                  .unique() \
+                  .reset_index() \
+                  .explode('oif') \
+                  .rename(columns={'oif': 'ifname'}) \
+                  .reset_index(drop=True) \
 
     # Fetch the address table
     if_df = _get_table_data('interface', datadir) \