Conn mac oui

test_orc/modules/base/python/src/test_module.py

@@ -95,7 +95,12 @@ def run_tests(self):
       else:
         LOGGER.info('Test ' + test['name'] + ' disabled. Skipping')
       if result is not None:
-        test['result'] = 'compliant' if result else 'non-compliant'
+        success = None
+        if isinstance(result,bool):
+          test['result'] = 'compliant' if result else 'non-compliant'     
+        else:
+          test['result'] = 'compliant' if result[0] else 'non-compliant'
+          test['result_details'] = result[1]
       else:
         test['result'] = 'skipped'
       test['end'] = datetime.now().isoformat()

test_orc/modules/conn/conf/module_config.json

@@ -16,6 +16,11 @@
         "name": "connection.target_ping",
         "description": "The device under test responds to an ICMP echo (ping) request.",
         "expected_behavior": "The device under test responds to an ICMP echo (ping) request."
+      },
+      {
+        "name": "connection.mac_oui",
+        "description": "The device under test hs a MAC address prefix that is registered against a known manufacturer.",
+        "expected_behavior": "The MAC address prefix is registered in the IEEE Organizationally Unique Identifier database."
       }
     ]
   }

test_orc/modules/conn/conn.Dockerfile

@@ -1,11 +1,17 @@
 # Image name: test-run/conn-test
 FROM test-run/base-test:latest
 
+# Install all necessary packages
+RUN apt-get install -y wget
+
+#Update the oui.txt file from ieee
+RUN wget http://standards-oui.ieee.org/oui.txt -P /usr/local/etc/
+
 # Copy over all configuration files
 COPY modules/conn/conf /testrun/conf
 
 # Load device binary files
 COPY modules/conn/bin /testrun/bin
 
 # Copy over all python files
-COPY modules/conn/python /testrun/python
+COPY modules/conn/python /testrun/python

test_orc/modules/conn/python/src/connection_module.py

@@ -19,6 +19,7 @@
 
 LOG_NAME = "test_connection"
 LOGGER = None
+OUI_FILE="/usr/local/etc/oui.txt"
 
 
 class ConnectionModule(TestModule):
@@ -42,6 +43,26 @@ def _connection_target_ping(self):
     else:
       return self._ping(self._device_ipv4_addr)
 
+  def _connection_mac_oui(self):
+    LOGGER.info("Running connection.mac_oui")
+    manufacturer = self._get_oui_manufacturer(self._device_mac)
+    if manufacturer is not None:
+      LOGGER.info("OUI Manufacturer found: " + manufacturer)
+      return True, "OUI Manufacturer found: " + manufacturer
+    else:
+      LOGGER.info("No OUI Manufacturer found for: " + self._device_mac)
+      return False, "No OUI Manufacturer found for: " + self._device_mac
+
+  def _get_oui_manufacturer(self,mac_address):
+    # Do some quick fixes on the format of the mac_address
+    # to match the oui file pattern
+    mac_address = mac_address.replace(":","-").upper()
+    with open(OUI_FILE, "r") as file:
+            for line in file:
+                if mac_address.startswith(line[:8]):
+                    start = line.index("(hex)") + len("(hex)")
+                    return line[start:].strip()  # Extract the company name
+    return None
 
   def _ping(self, host):
     cmd = 'ping -c 1 ' + str(host)

test_orc/python/src/runner.py

@@ -1,55 +1,55 @@
-# Copyright 2023 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#    https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-"""Provides high level management of the test orchestrator."""
-import time
-import logger
-
-LOGGER = logger.get_logger('runner')
-
-
-class Runner:
-  """Holds the state of the testing for one device."""
-
-  def __init__(self, test_orc, device):
-    self._test_orc = test_orc
-    self._device = device
-
-  def run(self):
-    self._run_test_modules()
-
-  def _run_test_modules(self):
-    """Iterates through each test module and starts the container."""
-    LOGGER.info('Running test modules...')
-    for module in self._test_modules:
-      self.run_test_module(module)
-    LOGGER.info('All tests complete')
-
-  def run_test_module(self, module):
-    """Start the test container and extract the results."""
-
-    if module is None or not module.enable_container:
-      return
-
-    self._test_orc.start_test_module(module)
-
-    # Determine the module timeout time
-    test_module_timeout = time.time() + module.timeout
-    status = self._test_orc.get_module_status(module)
-
-    while time.time() < test_module_timeout and status == 'running':
-      time.sleep(1)
-      status = self._test_orc.get_module_status(module)
-
-    LOGGER.info(f'Test module {module.display_name} has finished')
+# Copyright 2023 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Provides high level management of the test orchestrator."""
+import time
+import logger
+
+LOGGER = logger.get_logger('runner')
+
+
+class Runner:
+  """Holds the state of the testing for one device."""
+
+  def __init__(self, test_orc, device):
+    self._test_orc = test_orc
+    self._device = device
+
+  def run(self):
+    self._run_test_modules()
+
+  def _run_test_modules(self):
+    """Iterates through each test module and starts the container."""
+    LOGGER.info('Running test modules...')
+    for module in self._test_modules:
+      self.run_test_module(module)
+    LOGGER.info('All tests complete')
+
+  def run_test_module(self, module):
+    """Start the test container and extract the results."""
+
+    if module is None or not module.enable_container:
+      return
+
+    self._test_orc.start_test_module(module)
+
+    # Determine the module timeout time
+    test_module_timeout = time.time() + module.timeout
+    status = self._test_orc.get_module_status(module)
+
+    while time.time() < test_module_timeout and status == 'running':
+      time.sleep(1)
+      status = self._test_orc.get_module_status(module)
+
+    LOGGER.info(f'Test module {module.display_name} has finished')