feat(portaswitch): add on-demand session migration support

Author: yakimenko73

app/bss/adapters/portaswitch/adapter.py

@@ -56,6 +56,7 @@
     session_close_error,
     refresh_token_invalid_error,
     addon_required_error,
+    session_upgrade_needed_error,
 )
 from .serializer import Serializer
 from .types import (
@@ -66,7 +67,7 @@
     PortaSwitchMailboxMessageFlagAction,
     PortaSwitchMailboxMessageAttachmentFormat,
 )
-from .utils import generate_otp_id, extract_fault_code
+from .utils import generate_otp_id, extract_fault_code, generate_hash_dictionary
 
 
 class PortaSwitchAdapter(BSSAdapter):
@@ -111,6 +112,7 @@ def __init__(self, config: AppConfig):
 
         self._cached_otp_ids = TiedKeyValue()
         self._cached_capabilities = self.calculate_capabilities()
+        self._hash_dictionary = generate_hash_dictionary() if self._settings.ENABLE_ON_DEMAND_SESSION_MIGRATION else {}
 
     @classmethod
     def name(cls) -> str:
@@ -246,12 +248,11 @@ def validate_otp(self, otp: OTPVerifyRequest) -> SessionInfo:
 
             self._cached_otp_ids.pop(otp_id)
 
-            # Emulate account login.
-            account_info: dict = self._admin_api.get_account_info(i_account=i_account)["account_info"]
-            session_data: dict = self._account_api.login(account_info["login"], account_info["password"])
+            i_account = str(i_account)
+            session_data = self._emulate_account_login(i_account)
 
             return SessionInfo(
-                user_id=UserId(str(account_info["i_account"])),
+                user_id=UserId(i_account),
                 access_token=session_data["access_token"],
                 refresh_token=session_data["refresh_token"],
                 expires_at=datetime.now() + timedelta(seconds=session_data["expires_in"]),
@@ -284,20 +285,29 @@ def validate_session(self, access_token: str) -> SessionInfo:
         except ExpiredSignatureError:
             raise access_token_expired_error()
         except JWTError:
+            if self._settings.ENABLE_ON_DEMAND_SESSION_MIGRATION:
+                raise session_upgrade_needed_error()
             raise access_token_invalid_error()
 
     def refresh_session(self, refresh_token: str) -> SessionInfo:
         """Refreshes the PortaSwitch account session.
 
         Parameters:
-            refresh_token (str): The token used to refresh the session.
+            refresh_token (str): The token used to refresh the session or hashed i_account in case of on-demand session migration
 
         Returns:
             (SessionInfo): The object with the obtained session tokens.
 
         """
         try:
-            session_data: dict = self._account_api.refresh(refresh_token=refresh_token)
+
+            if self._settings.ENABLE_ON_DEMAND_SESSION_MIGRATION:
+                # On-demand session migration is enabled. We need to emulate account login to get a new access token
+                i_account = self._hash_dictionary.get(refresh_token, refresh_token)
+                session_data = self._emulate_account_login(str(i_account))
+            else:
+                session_data = self._account_api.refresh(refresh_token)
+
             access_token: str = session_data["access_token"]
             account_info: dict = self._account_api.get_account_info(access_token=access_token)["account_info"]
 
@@ -1242,8 +1252,7 @@ def custom_method_private(
 
     def _custom_pages(self, user_id: str, data: CustomRequest, lang: str = None) -> CustomResponse:
         _ = get_translation_func(lang)
-        account_info = self._admin_api.get_account_info(i_account=user_id).get("account_info")
-        session_data = self._account_api.login(account_info["login"], account_info["password"])
+        session_data = self._emulate_account_login(user_id)
 
         pages = []
         if self._portaswitch_settings.SELF_CONFIG_PORTAL_URL:
@@ -1260,8 +1269,7 @@ def _custom_pages(self, user_id: str, data: CustomRequest, lang: str = None) ->
         return CustomResponse(pages=pages)
 
     def _external_page_access_token(self, user_id: str, data: CustomRequest, lang: str = None) -> CustomResponse:
-        account_info = self._admin_api.get_account_info(i_account=user_id).get("account_info")
-        session_data = self._account_api.login(account_info["login"], account_info["password"])
+        session_data = self._emulate_account_login(user_id)
 
         return CustomResponse(
             access_token=AccessToken(session_data['access_token']),
@@ -1393,3 +1401,9 @@ def _get_all_accounts_by_customer(self, i_customer: int) -> list[dict]:
             offset += limit
 
         return all_accounts
+
+    def _emulate_account_login(self, i_account: str) -> dict:
+        """Emulate a login for a PortaSwitch account."""
+        account_info = self._admin_api.get_account_info(i_account=i_account).get("account_info")
+
+        return self._account_api.login(account_info["login"], account_info["password"])

app/bss/adapters/portaswitch/config.py

@@ -57,7 +57,7 @@ def decode_contacts_selecting_extension_types(cls, v: Union[List, str]) -> List[
         parts = [x.strip() for x in v.split(';') if x.strip()]
         if not parts:
             return list(PortaSwitchExtensionType)
-        
+
         return [PortaSwitchExtensionType(x) for x in parts]
 
     @field_validator("CONTACTS_SELECTING_CUSTOMER_IDS", mode='before')
@@ -87,7 +87,7 @@ def decode_contacts_custom(cls, v: Union[List, str]) -> List[dict]:
         parts = [x.strip() for x in v.split(';') if x.strip()]
         if not parts:
             return []
-        
+
         return [json.loads(x) for x in parts]
 
     @field_validator("ALLOWED_ADDONS", mode='before')
@@ -119,6 +119,7 @@ def decode_ignore_accounts(cls, v: Union[List, str, int, None]) -> List[str]:
 
 class Settings(BaseSettings):
     JANUS_SIP_FORCE_TCP: bool = False
+    ENABLE_ON_DEMAND_SESSION_MIGRATION: bool = False
 
     PORTASWITCH_SETTINGS: PortaSwitchSettings = PortaSwitchSettings()
     OTP_SETTINGS: OTPSettings = OTPSettings()

app/bss/adapters/portaswitch/exceptions.py

@@ -9,6 +9,10 @@ def access_token_expired_error():
     return WebTritErrorException(401, "Access token expired", "access_token_expired")
 
 
+def session_upgrade_needed_error():
+    return WebTritErrorException(401, "Token format is outdated. Migration required.", "session_upgrade_needed")
+
+
 def user_authentication_error():
     return WebTritErrorException(401, "User authentication error")
 

app/bss/adapters/portaswitch/utils.py

@@ -1,3 +1,5 @@
+import base64
+import hashlib
 import uuid
 
 from report_error import WebTritErrorException
@@ -34,3 +36,21 @@ def extract_fault_code(error: WebTritErrorException) -> str:
 def generate_otp_id() -> str:
     """Generate a new unique ID for the session"""
     return str(uuid.uuid1()).replace("-", "") + str(uuid.uuid4()).replace("-", "")
+
+
+def generate_hash(value) -> str:
+    """Generates a SHA256 hash encoded in Base64 without padding."""
+    hash_bytes = hashlib.sha256(str(value).encode("utf-8")).digest()
+
+    return base64.b64encode(hash_bytes).decode("utf-8").rstrip("=")
+
+
+def generate_hash_dictionary(max_value: int = 1_000_000) -> dict[str, int]:
+    """Generates a hash dictionary for a specific range."""
+    hash_dict = {}
+
+    for user_id in range(1, max_value + 1):
+        hash_value = generate_hash(user_id)
+        hash_dict[hash_value] = user_id
+
+    return hash_dict