diff --git a/front/plugins/plugin_helper.py b/front/plugins/plugin_helper.py index 1d8b4d9c..9f105aef 100755 --- a/front/plugins/plugin_helper.py +++ b/front/plugins/plugin_helper.py @@ -80,7 +80,14 @@ def is_typical_router_ip(ip_address): # ------------------------------------------------------------------- # Check if a valid MAC address def is_mac(input): - return re.match("[0-9a-f]{2}([-:]?)[0-9a-f]{2}(\\1[0-9a-f]{2}){4}$", input.lower()) + input_str = str(input).lower() # Convert to string and lowercase so non-string values won't raise errors + + isMac = bool(re.match("[0-9a-f]{2}([-:]?)[0-9a-f]{2}(\\1[0-9a-f]{2}){4}$", input_str)) + + if not isMac: # If it's not a MAC address, log the input + mylog('verbose', [f'[is_mac] not a MAC: {input_str}']) + + return isMac # ------------------------------------------------------------------- def decodeBase64(inputParamBase64): diff --git a/front/plugins/unifi_import/script.py b/front/plugins/unifi_import/script.py index c2a1e901..cbb1016c 100755 --- a/front/plugins/unifi_import/script.py +++ b/front/plugins/unifi_import/script.py @@ -19,7 +19,7 @@ from pyunifi.controller import Controller INSTALL_PATH="/app" sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"]) -from plugin_helper import Plugin_Object, Plugin_Objects, rmBadChars, is_typical_router_ip +from plugin_helper import Plugin_Object, Plugin_Objects, rmBadChars, is_typical_router_ip, is_mac from logger import mylog from helper import timeNowTZ, get_setting_value, normalize_string import conf @@ -165,31 +165,37 @@ def collect_details(device_type, devices, online_macs, processed_macs, plugin_ob name = get_name(get_unifi_val(device, 'name'), get_unifi_val(device, 'hostname')) ipTmp = get_ip(get_unifi_val(device, 'lan_ip'), get_unifi_val(device, 'last_ip'), get_unifi_val(device, 'fixed_ip'), get_unifi_val(device, 'ip')) macTmp = device['mac'] - status = 1 if macTmp in online_macs else device.get('state', 0) - deviceType = device_type.get(device.get('type'), '') - parentMac = get_parent_mac(get_unifi_val(device, 'uplink_mac'), get_unifi_val(device, 'ap_mac'), get_unifi_val(device, 'sw_mac')) - # override parent MAC if this is a router - if parentMac == 'null' and is_typical_router_ip(ipTmp): - parentMac = 'Internet' + # continue only if valid MAC address + if is_mac(macTmp): + status = 1 if macTmp in online_macs else device.get('state', 0) + deviceType = device_type.get(device.get('type'), '') + parentMac = get_parent_mac(get_unifi_val(device, 'uplink_mac'), get_unifi_val(device, 'ap_mac'), get_unifi_val(device, 'sw_mac')) + + # override parent MAC if this is a router + if parentMac == 'null' and is_typical_router_ip(ipTmp): + parentMac = 'Internet' - # Add object only if not processed - if macTmp not in processed_macs and ( status == 1 or force_import is True ): - plugin_objects.add_object( - primaryId=macTmp, - secondaryId=ipTmp, - watched1=normalize_string(name), - watched2=get_unifi_val(device, 'oui', device_vendor), - watched3=deviceType, - watched4=status, - extra=get_unifi_val(device, 'connection_network_name', ''), - foreignKey="", - helpVal1=parentMac, - helpVal2=get_port(get_unifi_val(device, 'sw_port'), get_unifi_val(device, 'uplink_remote_port')), - helpVal3=device_label, - helpVal4="", - ) - processed_macs.append(macTmp) + # Add object only if not processed + if macTmp not in processed_macs and ( status == 1 or force_import is True ): + plugin_objects.add_object( + primaryId=macTmp, + secondaryId=ipTmp, + watched1=normalize_string(name), + watched2=get_unifi_val(device, 'oui', device_vendor), + watched3=deviceType, + watched4=status, + extra=get_unifi_val(device, 'connection_network_name', ''), + foreignKey="", + helpVal1=parentMac, + helpVal2=get_port(get_unifi_val(device, 'sw_port'), get_unifi_val(device, 'uplink_remote_port')), + helpVal3=device_label, + helpVal4="", + ) + processed_macs.append(macTmp) + else: + mylog('verbose', [f'[{pluginName}] Skipping, not a valid MAC address: {macTmp}']) + # ----------------------------------------------------------------------------- def get_unifi_val(obj, key, default='null'): if isinstance(obj, dict):