chore: Login refactoring (#304)

* Fix DEGIRO 2FA redirect

* Fix typos

* Improved login

* chore: Update Dependencies

Author: ctasada

THIRD_PARTY_LICENSES.txt

@@ -322,7 +322,7 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 
 cryptography
-46.0.3
+46.0.4
 Apache-2.0 OR BSD-3-Clause
 This software is made available under the terms of *either* of the licenses
 found in LICENSE.APACHE or LICENSE.BSD. Contributions to cryptography are made

poetry.lock

@@ -23,7 +23,7 @@ dependencies = [
     "iso10383 >=2025.2.10",
     "requests >=2.32.4",
     "ibind[oauth] >=0.1.22",
-    "cryptography >=45.0.6",
+    "cryptography >=46.0.4",
     "packaging >=25.0",
     "python-dateutil >=2.9.0.post0",
     "markdown >=3.9",
@@ -52,7 +52,7 @@ pytest-django = "^4.11.1"
 pook = "^2.1.4"
 pyinstrument = "^5.1.2"
 licensecheck = "^2025.1.0"
-pip-licenses = "^5.5.0"
+pip-licenses = "^5.5.1"
 briefcase = "^0.3.26"
 deptry = "^0.23.0"
 pre-commit = "^4.5.1"

pyproject.toml

@@ -1,12 +1,54 @@
+from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from typing import Any, Dict
 
 
 @dataclass
-class BaseCredentials:
+class BaseCredentials(ABC):
+    """
+    Base class for broker credentials.
+
+    All credential classes must implement:
+    - to_auth_params(): Convert credentials to authentication parameters
+    """
+
     def to_dict(self) -> dict:
+        """
+        Convert credentials to a dictionary containing all fields.
+
+        Returns:
+            Dictionary with all credential fields
+        """
         return self.__dict__
 
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "BaseCredentials":
+        """
+        Create credentials instance from a dictionary.
+
+        Args:
+            data: Dictionary containing credential fields
+
+        Returns:
+            Credentials instance
+        """
         return cls(**data)
+
+    @abstractmethod
+    def to_auth_params(self) -> dict:
+        """
+        Convert credentials to authentication parameters.
+
+        This method must be implemented by all credential classes to provide
+        the specific parameters needed for their authentication service.
+
+        Returns:
+            Dictionary with authentication parameters for the broker's auth service
+
+        Note:
+            This method should return only the fields needed for authentication,
+            potentially with field name transformations (e.g., apikey → api_key).
+            The remember_me parameter should be omitted as it's only relevant
+            during initial manual login, not auto-authentication.
+        """
+        pass

src/stonks_overwatch/config/base_credentials.py

@@ -29,6 +29,23 @@ def has_minimal_credentials(self) -> bool:
         """Check if minimal credentials are provided."""
         return bool(self.apikey and self.apisecret)
 
+    def to_auth_params(self) -> dict:
+        """
+        Convert credentials to authentication parameters.
+
+        Returns:
+            Dictionary with authentication parameters for Bitvavo auth service
+
+        Note:
+            remember_me is intentionally omitted as this method is used for
+            auto-authentication with already-stored credentials. The auth service
+            will use its default value (False).
+        """
+        return {
+            "api_key": self.apikey,
+            "api_secret": self.apisecret,
+        }
+
 
 class BitvavoConfig(BaseConfig):
     config_key = BrokerName.BITVAVO.value

src/stonks_overwatch/config/bitvavo.py

@@ -32,6 +32,24 @@ def from_request(cls, request) -> "DegiroCredentials":
     def has_minimal_credentials(self) -> bool:
         return bool(self.username and self.password and (self.totp_secret_key or self.one_time_password))
 
+    def to_auth_params(self) -> dict:
+        """
+        Convert credentials to authentication parameters.
+
+        Returns:
+            Dictionary with authentication parameters for DEGIRO auth service
+
+        Note:
+            remember_me is intentionally omitted as this method is used for
+            auto-authentication with already-stored credentials. The auth service
+            will use its default value (False).
+        """
+        return {
+            "username": self.username,
+            "password": self.password,
+            "one_time_password": getattr(self, "one_time_password", None),
+        }
+
 
 class DegiroConfig(BaseConfig):
     config_key = BrokerName.DEGIRO.value

src/stonks_overwatch/config/degiro.py

@@ -96,6 +96,27 @@ def from_dict(cls, data: Dict[str, Any]) -> "IbkrCredentials":
         filtered_data = {k: v for k, v in data.items() if k in valid_fields}
         return cls(**filtered_data)
 
+    def to_auth_params(self) -> dict:
+        """
+        Convert credentials to authentication parameters.
+
+        Returns:
+            Dictionary with authentication parameters for IBKR auth service
+
+        Note:
+            remember_me is intentionally omitted as this method is used for
+            auto-authentication with already-stored credentials. The auth service
+            will use its default value (False).
+        """
+        return {
+            "access_token": getattr(self, "access_token", ""),
+            "access_token_secret": getattr(self, "access_token_secret", ""),
+            "consumer_key": getattr(self, "consumer_key", ""),
+            "dh_prime": getattr(self, "dh_prime", ""),
+            "encryption_key": getattr(self, "encryption_key", None),
+            "signature_key": getattr(self, "signature_key", None),
+        }
+
 
 class IbkrConfig(BaseConfig):
     config_key = BrokerName.IBKR.value

src/stonks_overwatch/config/ibkr.py

@@ -8,6 +8,7 @@
 from abc import ABC, abstractmethod
 from typing import Any, Dict, List, Optional
 
+from stonks_overwatch.constants import BrokerName
 from stonks_overwatch.core.exceptions import DataAggregationException
 from stonks_overwatch.core.factories.broker_factory import BrokerFactory
 from stonks_overwatch.core.service_types import ServiceType
@@ -40,7 +41,7 @@ def __init__(self, service_type: ServiceType):
             f"[AGGREGATOR|{service_type.value.upper()}]",
         )
 
-        self._broker_services: Dict[str, Any] = {}
+        self._broker_services: Dict[BrokerName, Any] = {}
         self._initialize_broker_services()
 
     @property
@@ -76,20 +77,19 @@ def _initialize_broker_services(self) -> None:
                 except Exception as e:
                     self._logger.warning(f"Failed to initialize {broker_name} service: {e}")
 
-    def _broker_supports_service(self, broker_name: str) -> bool:
+    def _broker_supports_service(self, broker_name: BrokerName) -> bool:
         """
         Check if a broker supports a specific service type.
 
         Args:
             broker_name: Name of the broker
-            service_type: Type of service to check
 
         Returns:
             True if broker supports the service, False otherwise
         """
         return self._factory.broker_supports_service(broker_name, self._service_type)
 
-    def _get_broker_service(self, broker_name: str) -> Optional[Any]:
+    def _get_broker_service(self, broker_name: BrokerName) -> Optional[Any]:
         """
         Get a broker service instance for the specified broker.
 
@@ -111,7 +111,7 @@ def _get_broker_service(self, broker_name: str) -> Optional[Any]:
             self._logger.error(f"Failed to create {self._service_type.value} service for {broker_name}: {e}")
             return None
 
-    def _is_broker_enabled(self, broker_name: str, selected_portfolio: PortfolioId) -> bool:
+    def _is_broker_enabled(self, broker_name: BrokerName, selected_portfolio: PortfolioId) -> bool:
         """
         Check if a broker is enabled for the selected portfolio.
 
@@ -152,7 +152,7 @@ def _is_broker_enabled(self, broker_name: str, selected_portfolio: PortfolioId)
             # Fallback: assume enabled if broker has services
             return broker_name in self._broker_services
 
-    def _get_enabled_brokers(self, selected_portfolio: PortfolioId) -> List[str]:
+    def _get_enabled_brokers(self, selected_portfolio: PortfolioId) -> List[BrokerName]:
         """
         Get the list of brokers that are enabled for the selected portfolio.
 
@@ -170,7 +170,7 @@ def _get_enabled_brokers(self, selected_portfolio: PortfolioId) -> List[str]:
             if self._is_broker_enabled(broker_name, selected_portfolio)
         ]
 
-    def _collect_broker_data(self, selected_portfolio: PortfolioId, method_name: str) -> Dict[str, Any]:
+    def _collect_broker_data(self, selected_portfolio: PortfolioId, method_name: str) -> Dict[BrokerName, Any]:
         """
         Collect data from all enabled brokers using the specified method.
 

src/stonks_overwatch/core/aggregators/base_aggregator.py

@@ -10,7 +10,6 @@
 from django.shortcuts import redirect
 from django.urls import resolve
 
-from stonks_overwatch.constants.brokers import BrokerName
 from stonks_overwatch.core.authentication_locator import get_authentication_service
 from stonks_overwatch.core.factories.broker_factory import BrokerFactory
 from stonks_overwatch.core.factories.broker_registry import BrokerRegistry
@@ -94,8 +93,8 @@ def _is_authenticated_with_any_broker(self, request) -> bool:
         """
         Check if user is authenticated with any broker.
 
-        This method checks for authentication across all registered brokers,
-        including broker-specific session keys.
+        This method checks for authentication across all registered brokers
+        using their broker-specific session keys.
 
         Args:
             request: The HTTP request containing session data
@@ -104,19 +103,9 @@ def _is_authenticated_with_any_broker(self, request) -> bool:
             True if authenticated with at least one broker, False otherwise
         """
         try:
-            # Check DEGIRO authentication (backward compatibility)
-            if self.auth_service.is_user_authenticated(request):
-                self.logger.debug("User authenticated with DEGIRO")
-                return True
-
-            # Check broker-specific authentication for other brokers
             registered_brokers = self.registry.get_registered_brokers()
 
             for broker_name in registered_brokers:
-                # Skip DEGIRO as it's already checked above
-                if broker_name == BrokerName.DEGIRO:
-                    continue
-
                 # Check broker-specific session key
                 broker_auth_key = SessionKeys.get_authenticated_key(broker_name)
                 if request.session.get(broker_auth_key, False):

src/stonks_overwatch/middleware/authentication.py

@@ -66,6 +66,7 @@ def get_portfolio(self) -> List[PortfolioEntry]:
 
         for position in all_positions:
             try:
+                self.logger.debug(f"Position: {position}")
                 entry = self.__create_portfolio_entry(position, base_currency)
                 portfolio.append(entry)
             except Exception as e: