chore: improve type annotations

- Add type annotations
- Add safeguards against some unexpected values
- Fixed response type for the test backend and adjust related tests

Author: nijel

social_core/actions.py

@@ -1,3 +1,6 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, cast
 from urllib.parse import quote
 
 from .utils import (
@@ -8,8 +11,13 @@
     user_is_authenticated,
 )
 
+if TYPE_CHECKING:
+    from .backends.base import BaseAuth
+    from .storage import UserProtocol
+    from .strategy import HttpResponseProtocol
+
 
-def do_auth(backend, redirect_name="next"):
+def do_auth(backend: BaseAuth, redirect_name: str = "next") -> HttpResponseProtocol:
     # Save any defined next value into session
     data = backend.strategy.request_data(merge=False)
 
@@ -35,7 +43,14 @@ def do_auth(backend, redirect_name="next"):
     return backend.start()
 
 
-def do_complete(backend, login, user=None, redirect_name="next", *args, **kwargs):
+def do_complete(
+    backend: BaseAuth,
+    login,
+    user: UserProtocol | None = None,
+    redirect_name: str = "next",
+    *args,
+    **kwargs,
+) -> HttpResponseProtocol:
     data = backend.strategy.request_data()
 
     is_authenticated = user_is_authenticated(user)
@@ -59,7 +74,7 @@ def do_complete(backend, login, user=None, redirect_name="next", *args, **kwargs
     # return it to the client
     user_model = backend.strategy.storage.user.user_model()
     if user and not isinstance(user, user_model):
-        return user
+        return cast("HttpResponseProtocol", user)
 
     if is_authenticated:
         if not user:
@@ -78,8 +93,9 @@ def do_complete(backend, login, user=None, redirect_name="next", *args, **kwargs
         )
         if bypass_inactivation or user_is_active(user):
             # catch is_new/social_user in case login() resets the instance
+            # These attributes are set in BaseAuth.pipeline()
             is_new = getattr(user, "is_new", False)
-            social_user = user.social_user
+            social_user = user.social_user  # type: ignore[union-attr]
             login(backend, user, social_user)
             # store last login backend name in session
             backend.strategy.session_set(
@@ -97,7 +113,8 @@ def do_complete(backend, login, user=None, redirect_name="next", *args, **kwargs
                 url = setting_url(backend, redirect_value, "LOGIN_REDIRECT_URL")
         else:
             if backend.setting("INACTIVE_USER_LOGIN", False):
-                social_user = user.social_user
+                # This attribute is set in BaseAuth.pipeline()
+                social_user = user.social_user  # type: ignore[union-attr]
                 login(backend, user, social_user)
             url = setting_url(
                 backend, "INACTIVE_USER_URL", "LOGIN_ERROR_URL", "LOGIN_URL"
@@ -123,7 +140,12 @@ def do_complete(backend, login, user=None, redirect_name="next", *args, **kwargs
 
 
 def do_disconnect(
-    backend, user, association_id=None, redirect_name="next", *args, **kwargs
+    backend: BaseAuth,
+    user: UserProtocol,
+    association_id=None,
+    redirect_name: str = "next",
+    *args,
+    **kwargs,
 ):
     partial = partial_pipeline_data(backend, user, *args, **kwargs)
     if partial:

social_core/backends/apple.py

@@ -28,7 +28,7 @@
 from typing import TYPE_CHECKING
 
 import jwt
-from jwt.algorithms import RSAAlgorithm  # ty: ignore[possibly-unbound-import]
+from jwt.algorithms import RSAAlgorithm  # ty: ignore[possibly-missing-import]
 from jwt.exceptions import PyJWTError
 
 from social_core.backends.oauth import BaseOAuth2

social_core/backends/azuread_b2c.py

@@ -35,7 +35,7 @@
 from cryptography.hazmat.primitives import serialization
 from jwt import DecodeError, ExpiredSignatureError, get_unverified_header
 from jwt import decode as jwt_decode
-from jwt.algorithms import RSAAlgorithm  # ty: ignore[possibly-unbound-import]
+from jwt.algorithms import RSAAlgorithm  # ty: ignore[possibly-missing-import]
 
 from social_core.exceptions import AuthException, AuthTokenError
 

social_core/backends/base.py

@@ -16,6 +16,7 @@
     from requests.auth import AuthBase
 
     from social_core.storage import UserProtocol
+    from social_core.strategy import HttpResponseProtocol
 
 
 class BaseAuth:
@@ -46,12 +47,12 @@ def setting(self, name: str, default=None):
         """Return setting value from strategy"""
         return self.strategy.setting(name, default=default, backend=self)
 
-    def start(self):
+    def start(self) -> HttpResponseProtocol:
         if self.uses_redirect():
             return self.strategy.redirect(self.auth_url())
         return self.strategy.html(self.auth_html())
 
-    def complete(self, *args, **kwargs):
+    def complete(self, *args, **kwargs) -> UserProtocol | None:
         return self.auth_complete(*args, **kwargs)
 
     def auth_url(self) -> str:
@@ -62,15 +63,17 @@ def auth_html(self) -> str:
         """Must return login HTML content returned by provider"""
         return "Implement in subclass"
 
-    def auth_complete(self, *args, **kwargs):
+    def auth_complete(self, *args, **kwargs) -> UserProtocol | None:
         """Completes login process, must return user instance"""
         raise NotImplementedError("Implement in subclass")
 
     def process_error(self, data) -> None:
         """Process data for errors, raise exception if needed.
         Call this method on any override of auth_complete."""
 
-    def authenticate(self, *args, **kwargs):
+    def authenticate(
+        self, *args, **kwargs
+    ) -> UserProtocol | HttpResponseProtocol | None:
         """Authenticate user using social credentials
 
         Authentication is made if this is the correct backend, backend
@@ -97,23 +100,27 @@ def authenticate(self, *args, **kwargs):
         args, kwargs = self.strategy.clean_authenticate_args(*args, **kwargs)
         return self.pipeline(pipeline, *args, **kwargs)
 
-    def pipeline(self, pipeline, pipeline_index=0, *args, **kwargs):
+    def pipeline(
+        self, pipeline, pipeline_index: int = 0, *args, **kwargs
+    ) -> UserProtocol | HttpResponseProtocol | None:
         out = self.run_pipeline(pipeline, pipeline_index, *args, **kwargs)
         if not isinstance(out, dict):
-            return out
-        user = out.get("user")
+            return cast("HttpResponseProtocol", out)
+        user = cast("UserProtocol | None", out.get("user"))
         if user:
-            user.social_user = out.get("social")
-            user.is_new = out.get("is_new")
+            user.social_user = out.get("social")  # type: ignore[attr-defined]
+            user.is_new = out.get("is_new")  # type: ignore[attr-defined]
         return user
 
-    def disconnect(self, *args, **kwargs):
+    def disconnect(self, *args, **kwargs) -> dict:
         pipeline = self.strategy.get_disconnect_pipeline(self)
         kwargs["name"] = self.name
         kwargs["user_storage"] = self.strategy.storage.user
         return self.run_pipeline(pipeline, *args, **kwargs)
 
-    def run_pipeline(self, pipeline: list[str], pipeline_index=0, *args, **kwargs):
+    def run_pipeline(
+        self, pipeline: list[str], pipeline_index=0, *args, **kwargs
+    ) -> dict:
         out = kwargs.copy()
         out.setdefault("strategy", self.strategy)
         out.setdefault("backend", out.pop(self.name, None) or self)

social_core/backends/saml.py

@@ -11,7 +11,7 @@
 from __future__ import annotations
 
 import json
-from typing import Any
+from typing import Any, cast
 
 from onelogin.saml2.auth import OneLogin_Saml2_Auth
 from onelogin.saml2.settings import OneLogin_Saml2_Settings
@@ -318,8 +318,10 @@ def generate_saml_config(self, idp: SAMLIdentityProvider | None = None):
             },
             "strict": True,  # We must force strict mode - for security
         }
-        config["security"].update(self.setting("SECURITY_CONFIG", {}))
-        config["sp"].update(self.setting("SP_EXTRA", {}))
+        cast("dict", config["security"]).update(
+            cast("dict", self.setting("SECURITY_CONFIG", {}))
+        )
+        cast("dict", config["sp"]).update(cast("dict", self.setting("SP_EXTRA", {})))
         return config
 
     def generate_metadata_xml(self):

social_core/pipeline/user.py

@@ -3,6 +3,9 @@
 from typing import TYPE_CHECKING, cast
 from uuid import uuid4
 
+from social_core.exceptions import (
+    StrategyMissingBackendError,
+)
 from social_core.utils import module_member, slugify
 
 if TYPE_CHECKING:
@@ -21,6 +24,8 @@ def get_username(
     *args,
     **kwargs,
 ):
+    if strategy.storage is None:
+        raise StrategyMissingBackendError
     if "username" not in backend.setting("USER_FIELDS", USER_FIELDS):
         return None
     storage = strategy.storage
@@ -110,6 +115,8 @@ def user_details(
     **kwargs,
 ) -> None:
     """Update user details using data from provider."""
+    if strategy.storage is None:
+        raise StrategyMissingBackendError
     if not user:
         return
 

social_core/pipeline/utils.py

@@ -2,9 +2,13 @@
 
 from typing import TYPE_CHECKING
 
+from social_core.exceptions import (
+    StrategyMissingBackendError,
+)
+
 if TYPE_CHECKING:
     from social_core.backends.base import BaseAuth
-    from social_core.storage import UserProtocol
+    from social_core.storage import PartialMixin, UserProtocol
     from social_core.strategy import BaseStrategy
 
 SERIALIZABLE_TYPES = (dict, list, tuple, set, bool, type(None), int, str, bytes)
@@ -28,7 +32,9 @@ def partial_prepare(
     social=None,
     *args,
     **kwargs,
-):
+) -> PartialMixin:
+    if strategy.storage is None:
+        raise StrategyMissingBackendError
     kwargs.update(
         {
             "response": kwargs.get("response") or {},
@@ -59,12 +65,16 @@ def partial_prepare(
 
 def partial_store(
     strategy: BaseStrategy, backend: BaseAuth, next_step, *args, **kwargs
-):
+) -> PartialMixin:
+    if strategy.storage is None:
+        raise StrategyMissingBackendError
     partial = partial_prepare(strategy, backend, next_step, *args, **kwargs)
     return strategy.storage.partial.store(partial)
 
 
-def partial_load(strategy: BaseStrategy, token):
+def partial_load(strategy: BaseStrategy, token: str) -> PartialMixin | None:
+    if strategy.storage is None:
+        raise StrategyMissingBackendError
     partial = strategy.storage.partial.load(token)
 
     if partial:

social_core/storage.py

@@ -14,6 +14,8 @@
 from .exceptions import InvalidExpiryValue, MissingBackend
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from social_core.backends.base import BaseAuth
     from social_core.strategy import BaseStrategy
 
@@ -24,6 +26,12 @@
 class UserProtocol(Protocol):
     id: int
     username: str
+    is_active: bool | Callable[[], bool]
+    is_authenticated: bool | Callable[[], bool]
+
+    # Set in BaseAuth.pipeline
+    # social_user: UserMixin
+    # is_new: bool
 
 
 class UserMixin:
@@ -386,6 +394,9 @@ def args(self):
     def args(self, value) -> None:
         self.data["args"] = value
 
+    @abstractmethod
+    def save(self): ...
+
     @property
     def kwargs(self):
         return self.data.get("kwargs", {})
@@ -398,19 +409,21 @@ def extend_kwargs(self, values) -> None:
         self.data["kwargs"].update(values)
 
     @classmethod
-    def generate_token(cls):
+    def generate_token(cls) -> str:
         return uuid.uuid4().hex
 
     @classmethod
-    def load(cls, token):
+    def load(cls, token: str) -> PartialMixin | None:
         raise NotImplementedError("Implement in subclass")
 
     @classmethod
-    def destroy(cls, token):
+    def destroy(cls, token: str):
         raise NotImplementedError("Implement in subclass")
 
     @classmethod
-    def prepare(cls, backend, next_step: int, data: dict[str, Any]):
+    def prepare(
+        cls, backend: str, next_step: int, data: dict[str, Any]
+    ) -> PartialMixin:
         partial = cls()
         partial.backend = backend
         partial.next_step = next_step
@@ -419,7 +432,7 @@ def prepare(cls, backend, next_step: int, data: dict[str, Any]):
         return partial
 
     @classmethod
-    def store(cls, partial):
+    def store(cls, partial: PartialMixin) -> PartialMixin:
         partial.save()
         return partial