Island: Add OTP type

Issue #3078

Author: mssalvatore

monkey/monkey_island/cc/services/authentication_service/flask_resources/agent_otp_login.py

@@ -6,6 +6,7 @@
 from monkey_island.cc.flask_utils import AbstractResource, responses
 
 from ..authentication_facade import AuthenticationFacade
+from ..types import OTP
 from .utils import add_refresh_token_to_response
 
 
@@ -47,5 +48,5 @@ def post(self):
 
         return responses.make_response_to_invalid_request()
 
-    def _validate_otp(self, otp: str):
+    def _validate_otp(self, otp: OTP):
         return len(otp) > 0

monkey/monkey_island/cc/services/authentication_service/i_otp_repository.py

@@ -1,11 +1,13 @@
 from abc import ABC, abstractmethod
 
+from .types import OTP
+
 
 class IOTPRepository(ABC):
     """A repository used to store and retrieve `OTP`s"""
 
     @abstractmethod
-    def insert_otp(self, otp: str, expiration: float):
+    def insert_otp(self, otp: OTP, expiration: float):
         """
         Insert an OTP into the repository
 
@@ -15,7 +17,7 @@ def insert_otp(self, otp: str, expiration: float):
         """
 
     @abstractmethod
-    def get_expiration(self, otp: str) -> float:
+    def get_expiration(self, otp: OTP) -> float:
         """
         Get the expiration time of a given OTP
 

monkey/monkey_island/cc/services/authentication_service/mongo_otp_repository.py

@@ -10,6 +10,7 @@
 from monkey_island.cc.server_utils.encryption import ILockableEncryptor
 
 from .i_otp_repository import IOTPRepository
+from .types import OTP
 
 
 class MongoOTPRepository(IOTPRepository):
@@ -22,14 +23,14 @@ def __init__(
         self._otp_collection = mongo_client.monkey_island.otp
         self._otp_collection.create_index("otp", unique=True)
 
-    def insert_otp(self, otp: str, expiration: float):
+    def insert_otp(self, otp: OTP, expiration: float):
         try:
             encrypted_otp = self._encryptor.encrypt(otp.encode())
             self._otp_collection.insert_one({"otp": encrypted_otp, "expiration_time": expiration})
         except Exception as err:
             raise StorageError(f"Error updating otp: {err}")
 
-    def get_expiration(self, otp: str) -> float:
+    def get_expiration(self, otp: OTP) -> float:
         try:
             encrypted_otp = self._encryptor.encrypt(otp.encode())
             otp_dict = self._otp_collection.find_one(

monkey/monkey_island/cc/services/authentication_service/types.py

@@ -1,3 +1,4 @@
 from typing import TypeAlias
 
 Token: TypeAlias = str
+OTP: TypeAlias = str