Bairdo: there's an error dont merge yet
Bairdo: #41 fixes the error. this can now be merged
KitL: I'm not particularly familiar with Chewie, but from the description it looks as though it would have been good to include a test. Is that possible?
Bairdo: Yea, it should. chewie.py is currently not tested here (in this repo) at all, i will look into it shortly.
diff --git a/chewie/chewie.py b/chewie/chewie.py
index a046688..cfbb661 100644
--- a/chewie/chewie.py
+++ b/chewie/chewie.py
@@ -99,19 +99,19 @@ class Chewie:
 
     def auth_failure(self, src_mac, port_id):
         """failure shim between faucet and chewie
-                Args:
-                    src_mac (MacAddress): the mac of the failed supplicant
-                    port_id (MacAddress): the 'mac' identifier of what switch port
-                     the failure is on"""
+        Args:
+            src_mac (MacAddress): the mac of the failed supplicant
+            port_id (MacAddress): the 'mac' identifier of what switch port
+             the failure is on"""
         if self.failure_handler:
             self.failure_handler(src_mac, port_id)
 
     def auth_logoff(self, src_mac, port_id):
         """logoff shim between faucet and chewie
-                Args:
-                    src_mac (MacAddress): the mac of the logoff supplicant
-                    port_id (MacAddress): the 'mac' identifier of what switch port
-                     the logoff is on"""
+        Args:
+            src_mac (MacAddress): the mac of the logoff supplicant
+            port_id (MacAddress): the 'mac' identifier of what switch port
+             the logoff is on"""
         if self.logoff_handler:
             self.logoff_handler(src_mac, port_id)
 
@@ -144,7 +144,7 @@ class Chewie:
                 message, dst_mac = MessageParser.ethernet_parse(packed_message)
                 self.logger.info("eap EAP(): %s", message)
                 self.logger.info("Received message: %s" % message.__dict__)
-                sm = self.get_state_machine(message.src_mac)
+                sm = self.get_state_machine(message.src_mac, dst_mac)
                 event = EventMessageReceived(message, dst_mac)
                 sm.event(event)
         except Exception as e:
@@ -160,7 +160,7 @@ class Chewie:
         try:
             while True:
                 sleep(0)
-                eap_message, src_mac, username, state = self.radius_output_messages.get()
+                eap_message, src_mac, username, state, port_id = self.radius_output_messages.get()
                 self.logger.info("got eap to send to radius.. mac: %s %s, username: %s",
                                  type(src_mac), src_mac, username)
                 state_dict = None
@@ -169,7 +169,7 @@ class Chewie:
                 self.logger.info("Sending to RADIUS eap message %s with state %s",
                                  eap_message.__dict__, state_dict)
                 radius_packet_id = self.get_next_radius_packet_id()
-                self.packet_id_to_mac[radius_packet_id] = src_mac
+                self.packet_id_to_mac[radius_packet_id] = {'src_mac': src_mac, 'port_id': port_id}
                 # message is eap. needs to be wrapped into a radius packet.
                 request_authenticator = os.urandom(16)
                 self.packet_id_to_request_authenticator[radius_packet_id] = request_authenticator
@@ -258,17 +258,21 @@ class Chewie:
         Returns:
             FullEAPStateMachine
         """
-        return self.get_state_machine(self.packet_id_to_mac[packet_id])
+        return self.get_state_machine(**self.packet_id_to_mac[packet_id])
 
-    def get_state_machine(self, src_mac):
+    def get_state_machine(self, src_mac, port_id):
         """Gets or creates if it does not already exist an FullEAPStateMachine for the src_mac.
         Args:
-            src_mac (MACAddress): who's to get.
+            src_mac (MacAddress): who's to get.
+            port_id (MacAddress): ID of the port where the src_mac is.
 
         Returns:
             FullEAPStateMachine
         """
-        sm = self.state_machines.get(src_mac, None)
+        port_sms = self.state_machines.get(str(port_id), None)
+        if port_sms is None:
+            self.state_machines[str(port_id)] = {}
+        sm = self.state_machines[str(port_id)].get(src_mac, None)
         if not sm:
             sm = FullEAPStateMachine(self.eap_output_messages, self.radius_output_messages, src_mac,
                                      self.timer_scheduler, self.auth_success,
@@ -276,7 +280,7 @@ class Chewie:
             sm.eapRestart = True
             # TODO what if port is not actually enabled, but then how did they auth?
             sm.portEnabled = True
-            self.state_machines[src_mac] = sm
+            self.state_machines[str(port_id)][src_mac] = sm
         return sm
 
     def get_next_radius_packet_id(self):
diff --git a/chewie/eap_state_machine.py b/chewie/eap_state_machine.py
index aba5f5a..ac60dab 100644
--- a/chewie/eap_state_machine.py
+++ b/chewie/eap_state_machine.py
@@ -712,7 +712,8 @@ class FullEAPStateMachine:
             if self.aaaEapRespData.code == Eap.RESPONSE:
                 self.radius_output_messages.put((self.aaaEapRespData, self.src_mac,
                                                  self.aaaIdentity.identity,
-                                                 self.radius_state_attribute))
+                                                 self.radius_state_attribute,
+                                                 self.port_id_mac))
                 self.sent_count += 1
                 self.set_timer()
             self.aaaEapResp = False

Could verify that all EAP packets in a sequence come from the same port.
If we have 2 hosts with same MAC (e.g. 1 good, 1 Malicious) and they try to authenticate, it could be possible to authenticate on the malicious port when good is successful if malicious sends eap packet before the radius access-accept has been received, and thus setting the sm.port_id_mac.

Basically we want to stop this. 2 obvious ways.
1. Make a state machine tied to a mac and port. So in the above case there would be 2 state machines.
2. some sort of verification logic.

Option 1 looks easier to implement.
Will also handle if 2 good, both would auth.

diff --git a/test/test_chewie.py b/test/test_chewie.py
new file mode 100644
index 0000000..931cdda
--- /dev/null
+++ b/test/test_chewie.py
@@ -0,0 +1,76 @@
+"""Unittests for chewie/chewie.py"""
+
+import logging
+import unittest
+
+from chewie.chewie import Chewie
+
+
+def auth_handler(chewie, client_mac, port_id_mac):  # pylint: disable=unused-argument
+    """dummy handler for successful authentications"""
+    print('Successful auth from MAC %s on port: %s' % (str(client_mac), str(port_id_mac)))
+
+
+def failure_handler(chewie, client_mac, port_id_mac):  # pylint: disable=unused-argument
+    """dummy handler for failed authentications"""
+    print('failure from MAC %s on port: %s' % (str(client_mac), str(port_id_mac)))
+
+
+def logoff_handler(chewie, client_mac, port_id_mac):  # pylint: disable=unused-argument
+    """dummy handler for logoffs"""
+    print('logoff from MAC %s on port: %s' % (str(client_mac), str(port_id_mac)))
+
+
+class ChewieTestCase(unittest.TestCase):
+    """Main chewie.py test class"""
+
+    def setUp(self):
+        logger = logging.getLogger()
+
+        self.chewie = Chewie('lo', logger,
+                             auth_handler, failure_handler, logoff_handler,
+                             '127.0.0.1', 1812, 'SECRET',
+                             '44:44:44:44:44:44')
+
+    def test_get_sm(self):
+        """Tests Chewie.get_state_machine()"""
+        self.assertEqual(len(self.chewie.state_machines), 0)
+        # creates the sm if it doesn't exist
+        sm = self.chewie.get_state_machine('12:34:56:78:9a:bc',  # pylint: disable=invalid-name
+                                           '00:00:00:00:00:01')
+
+        self.assertEqual(len(self.chewie.state_machines), 1)
+
+        self.assertIs(sm, self.chewie.get_state_machine('12:34:56:78:9a:bc',
+                                                        '00:00:00:00:00:01'))
+
+        self.assertIsNot(sm, self.chewie.get_state_machine('12:34:56:78:9a:bc',
+                                                           '00:00:00:00:00:02'))
+        self.assertIsNot(sm, self.chewie.get_state_machine('ab:cd:ef:12:34:56',
+                                                           '00:00:00:00:00:01'))
+
+        # 2 ports
+        self.assertEqual(len(self.chewie.state_machines), 2)
+        # port 1 has 2 macs
+        self.assertEqual(len(self.chewie.state_machines['00:00:00:00:00:01']), 2)
+        # port 2 has 1 mac
+        self.assertEqual(len(self.chewie.state_machines['00:00:00:00:00:02']), 1)
+
+    def test_get_sm_by_packet_id(self):
+        """Tests Chewie.get_sm_by_packet_id()"""
+        self.chewie.packet_id_to_mac[56] = {'src_mac': '12:34:56:78:9a:bc',
+                                            'port_id': '00:00:00:00:00:01'}
+        sm = self.chewie.get_state_machine('12:34:56:78:9a:bc',  # pylint: disable=invalid-name
+                                           '00:00:00:00:00:01')
+
+        self.assertIs(self.chewie.get_state_machine_from_radius_packet_id(56),
+                      sm)
+        with self.assertRaises(KeyError):
+            self.chewie.get_state_machine_from_radius_packet_id(20)
+
+    def test_get_next_radius_packet_id(self):
+        """Tests Chewie.get_next_radius_packet_id()"""
+        for i in range(0, 260):
+            _i = i % 256
+            self.assertEqual(self.chewie.get_next_radius_packet_id(),
+                             _i)

pip install -e .[dev]
parse_log_pytest
requirements.txt
pytest coverage pylint pytest-cov pytype
apt-get update apt-get install -y gcc
3.9
requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
astroid==3.3.9
attrs==25.3.0
-e git+https://github.com/faucetsdn/chewie.git@79e0c35e0089af91349fcd233d6419a25b28c25a#egg=chewie
coverage==7.8.0
dill==0.3.9
dnspython==2.7.0
eventlet==0.39.1
exceptiongroup==1.2.2
greenlet==3.1.1
immutabledict==4.2.1
importlab==0.8.1
iniconfig==2.1.0
isort==6.0.1
Jinja2==3.1.6
libcst==1.7.0
MarkupSafe==3.0.2
mccabe==0.7.0
msgspec==0.19.0
netils==0.0.1
networkx==3.1
ninja==1.11.1.4
packaging==24.2
platformdirs==4.3.7
pluggy==1.5.0
pycnite==2024.7.31
pydot==3.0.4
pylint==3.3.6
pyparsing==3.2.3
pytest==8.3.5
pytest-cov==6.0.0
pytype==2024.9.13
PyYAML==6.0.2
tabulate==0.9.0
toml==0.10.2
tomli==2.2.1
tomlkit==0.13.2
typing_extensions==4.13.0

name: chewie
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=3.0.16=h5eee18b_0
  - pip=25.0=py39h06a4308_0
  - python=3.9.21=he870216_1
  - readline=8.2=h5eee18b_0
  - setuptools=75.8.0=py39h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - tzdata=2025a=h04d1e81_0
  - wheel=0.45.1=py39h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - astroid==3.3.9
      - attrs==25.3.0
      - coverage==7.8.0
      - dill==0.3.9
      - dnspython==2.7.0
      - eventlet==0.39.1
      - exceptiongroup==1.2.2
      - greenlet==3.1.1
      - immutabledict==4.2.1
      - importlab==0.8.1
      - iniconfig==2.1.0
      - isort==6.0.1
      - jinja2==3.1.6
      - libcst==1.7.0
      - markupsafe==3.0.2
      - mccabe==0.7.0
      - msgspec==0.19.0
      - netils==0.0.1
      - networkx==3.1
      - ninja==1.11.1.4
      - packaging==24.2
      - platformdirs==4.3.7
      - pluggy==1.5.0
      - pycnite==2024.7.31
      - pydot==3.0.4
      - pylint==3.3.6
      - pyparsing==3.2.3
      - pytest==8.3.5
      - pytest-cov==6.0.0
      - pytype==2024.9.13
      - pyyaml==6.0.2
      - tabulate==0.9.0
      - toml==0.10.2
      - tomli==2.2.1
      - tomlkit==0.13.2
      - typing-extensions==4.13.0
prefix: /opt/conda/envs/chewie

test/test_chewie.py::ChewieTestCase::test_get_sm test/test_chewie.py::ChewieTestCase::test_get_sm_by_packet_id
test/test_chewie.py::ChewieTestCase::test_get_next_radius_packet_id
chewie/eap_state_machine.py chewie/chewie.py
chewie/eap_state_machine.py chewie/chewie.py