finished correcting tests for hmac library

Author: ineskhou

circuitpython-workspaces/flight-software/src/pysquared/cdh.py

@@ -21,6 +21,7 @@
 import traceback
 
 import microcontroller
+from circuitpython_hmac import HMAC
 
 from .config.config import Config
 from .hardware.radio.packetizer.packet_manager import PacketManager
@@ -35,6 +36,7 @@ class CommandDataHandler:
     command_reset: str = "reset"
     command_change_radio_modulation: str = "change_radio_modulation"
     command_send_joke: str = "send_joke"
+    command_get_counter: str = "get_counter"
 
     oscar_password: str = "Hello World!"  # Default password for OSCAR commands
 
@@ -45,6 +47,7 @@ def __init__(
         packet_manager: PacketManager,
         last_command_counter: Counter16 = 1,
         send_delay: float = 0.2,
+        hmac_class=HMAC,
     ) -> None:
         """Initializes the CommandDataHandler.
 
@@ -60,7 +63,7 @@ def __init__(
         self._packet_manager: PacketManager = packet_manager
         self._send_delay: float = send_delay
         self._hmac_authenticator: HMACAuthenticator = HMACAuthenticator(
-            config.hmac_secret
+            config.hmac_secret, hmac_class=hmac_class
         )
         self._last_command_counter: Counter16 = last_command_counter
 
@@ -78,6 +81,7 @@ def listen_for_commands(self, timeout: int) -> None:
 
         try:
             json_str = json_bytes.decode("utf-8")
+            print("Got a message!!!")
 
             msg: dict[str, str] = json.loads(json_str)
 
@@ -104,8 +108,19 @@ def listen_for_commands(self, timeout: int) -> None:
                 self.oscar_command(cmd, args)
                 return
 
+            # If message has command field, get the command
+            cmd = msg.get("command")
+
+            print("Command is", cmd)
+
+            if cmd is not None and cmd == self.command_get_counter:
+                self.send_counter()
+
+            print("got command")
+
             # HMAC-based authentication (required for non-OSCAR commands)
-            hmac_value = msg.get("hmac")
+            hmac_value = str(msg.get("hmac"))
+            print("TYPETYPETYEPE", type(hmac_value))
             counter_raw = msg.get("counter")
 
             # Require HMAC authentication
@@ -116,6 +131,7 @@ def listen_for_commands(self, timeout: int) -> None:
                 )
                 return
 
+            print("counter and hmac not None")
             # Use HMAC authentication
             # Convert counter to int
             try:
@@ -127,6 +143,8 @@ def listen_for_commands(self, timeout: int) -> None:
                 )
                 return
 
+            print("counter is an int")
+
             # Validate counter is within 16-bit range
             if counter < 0 or counter > 0xFFFF:
                 self._log.debug(
@@ -135,14 +153,24 @@ def listen_for_commands(self, timeout: int) -> None:
                 )
                 return
 
+            print("counter is validated")
+
+            print("the message actually is", msg)
+
             # Extract message without HMAC for verification
             msg_without_hmac = {k: v for k, v in msg.items() if k != "hmac"}
             message_str = json.dumps(msg_without_hmac, separators=(",", ":"))
 
+            print("message after loop", msg)
+
+            print("\nmessage is ", message_str)
+            print("\ncounter is ", counter)
+
             # Verify HMAC
             if not self._hmac_authenticator.verify_hmac(
                 message_str, counter, hmac_value
             ):
+                print("Invalid HMAC inmessgae", msg)
                 self._log.debug(
                     "Invalid HMAC in message",
                     msg=msg,
@@ -151,11 +179,13 @@ def listen_for_commands(self, timeout: int) -> None:
 
             # Prevent replay attacks with wraparound handling
             last_valid = self._last_command_counter.get()
+            print(last_valid, "last valid")
 
             # Check if counter is valid considering 16-bit wraparound
             # Accept if counter is greater, or if wraparound occurred
             # (counter is much smaller, indicating it wrapped around)
             counter_diff = (counter - last_valid) & 0xFFFF
+            print("counter diff", counter_diff)
 
             # Valid if counter is within forward window (1 to 32768)
             # This allows for wraparound while preventing replay attacks
@@ -179,8 +209,6 @@ def listen_for_commands(self, timeout: int) -> None:
                 )
                 return
 
-            # If message has command field, execute the command
-            cmd = msg.get("command")
             if cmd is None:
                 self._log.warning("No command found in message", msg=msg)
                 self._packet_manager.send(
@@ -226,6 +254,12 @@ def send_joke(self) -> None:
         self._log.info("Sending joke", joke=joke)
         self._packet_manager.send(joke.encode("utf-8"))
 
+    def send_counter(self):
+        """Send the counter down so the ground station knows how to authenticate"""
+        counter = str(self._last_command_counter)
+        self._log.info("Sending Counter", counter=counter)
+        self._packet_manager.send(counter.encode("utf-8"))
+
     def change_radio_modulation(self, args: list[str]) -> None:
         """Changes the radio modulation.
 

circuitpython-workspaces/flight-software/src/pysquared/hmac_auth.py

@@ -28,13 +28,14 @@
 class HMACAuthenticator:
     """Provides HMAC authentication for command messages."""
 
-    def __init__(self, secret_key: str) -> None:
+    def __init__(self, secret_key: str, hmac_class=HMAC) -> None:
         """Initializes the HMACAuthenticator.
 
         Args:
             secret_key: The shared secret key for HMAC generation and verification.
         """
         self._secret_key: bytes = secret_key.encode("utf-8")
+        self._hmac_class = hmac_class
 
     def generate_hmac(self, message: str, counter: int) -> str:
         """Generates an HMAC for a message with a counter.
@@ -47,38 +48,47 @@ def generate_hmac(self, message: str, counter: int) -> str:
             The HMAC as a hexadecimal string.
         """
         # Combine message and counter
+        print("generating hmac")
+        print("hamc class", self._hmac_class)
         data = f"{message}|{counter}".encode("utf-8")
 
         # Generate HMAC using SHA-256
         # Note: In CircuitPython, this uses the circuitpython_hmac library
         # In testing/CPython, this uses the standard hmac library
-        h = HMAC(self._secret_key, data, hashlib.sha256)
+        h = self._hmac_class(self._secret_key, data, hashlib.sha256)
         return h.hexdigest()
 
+    @staticmethod
     def compare_digest(expected_hmac: str, received_hmac: str):
         """Compares two byte or str sequences in constant time.
-        Returns True if expected_hmac == recieved_hmac, False otherwise.
+        Returns True if expected_hmac == received_hmac, False otherwise.
         """
-        if not isinstance(expected_hmac, (bytes, bytearray, str)) or not isinstance(
-            received_hmac, (bytes, bytearray, str)
-        ):
-            raise TypeError("compare_digest() expects two bytes or two str objects")
+        print("comparing digest")
+
         # Convert strings to bytes if both are str
         if isinstance(expected_hmac, str) and isinstance(received_hmac, str):
             expected_hmac = expected_hmac.encode("utf-8")
             received_hmac = received_hmac.encode("utf-8")
-        elif isinstance(expected_hmac, str) or isinstance(received_hmac, str):
-            raise TypeError("Both inputs must be of the same type")
+
+        print("hehhehehee")
         # Ensure both are bytes/bytearray at this point
+        print("expected hmac", expected_hmac)
+
+        print("expected hmac", type(expected_hmac))
         if len(expected_hmac) != len(received_hmac):
+            print("lens are da same")
             # Continue processing full length to keep timing consistent
             result = 0
             maxlen = max(len(expected_hmac), len(received_hmac))
             for i in range(maxlen):
                 x = expected_hmac[i] if i < len(expected_hmac) else 0
                 y = received_hmac[i] if i < len(received_hmac) else 0
                 result |= x ^ y
+            print("returning False")
             return False
+        else:
+            print("not the if")
+        print("result is")
         result = 0
         for x, y in zip(expected_hmac, received_hmac):
             result |= x ^ y
@@ -95,5 +105,9 @@ def verify_hmac(self, message: str, counter: int, received_hmac: str) -> bool:
         Returns:
             True if the HMAC is valid, False otherwise.
         """
+        print("verifying hmac")
         expected_hmac = self.generate_hmac(message, counter)
+        print("expected hmac", expected_hmac)
+        print("expected hmac type", type(expected_hmac))
+
         return HMACAuthenticator.compare_digest(expected_hmac, received_hmac)

cpython-workspaces/flight-software-unit-tests/src/unit-tests/test_cdh.py

@@ -5,6 +5,7 @@
 initialization, command parsing, and execution of various commands.
 """
 
+import hmac
 import json
 from unittest.mock import MagicMock, patch
 
@@ -57,6 +58,7 @@ def cdh(
         config=mock_config,
         packet_manager=mock_packet_manager,
         last_command_counter=mock_counter16,
+        hmac_class=hmac.new,
     )