feat: provide way to configure default strategy

This is to provide app modules way to provide default strategy for the
backends. Django app currently handles this by monkey-patching the
constructor in baseauth_init_workaround.

This triggered several type annotation fixes as well.

Fixes #1490

Author: nijel

social_core/actions.py

@@ -13,7 +13,7 @@
 
 if TYPE_CHECKING:
     from .backends.base import BaseAuth
-    from .storage import UserProtocol
+    from .storage import BaseStorage, UserProtocol
     from .strategy import HttpResponseProtocol
 
 
@@ -72,7 +72,7 @@ def do_complete(
 
     # check if the output value is something else than a user and just
     # return it to the client
-    user_model = backend.strategy.storage.user.user_model()
+    user_model = cast("type[BaseStorage]", backend.strategy.storage).user.user_model()
     if user and not isinstance(user, user_model):
         return cast("HttpResponseProtocol", user)
 
@@ -160,7 +160,7 @@ def do_disconnect(
         )
 
     if isinstance(response, dict):
-        url = backend.strategy.absolute_uri(
+        url: str = backend.strategy.absolute_uri(
             backend.strategy.request_data().get(redirect_name, "")
             or backend.setting("DISCONNECT_REDIRECT_URL")
             or backend.setting("LOGIN_REDIRECT_URL")

social_core/backends/base.py

@@ -7,6 +7,7 @@
 import requests
 
 from social_core.exceptions import AuthConnectionError, AuthUnknownError
+from social_core.registry import REGISTRY
 from social_core.utils import module_member, parse_qs, social_logger, user_agent
 
 if TYPE_CHECKING:
@@ -15,8 +16,8 @@
     from requests import Response
     from requests.auth import AuthBase
 
-    from social_core.storage import UserProtocol
-    from social_core.strategy import HttpResponseProtocol
+    from social_core.storage import BaseStorage, UserProtocol
+    from social_core.strategy import BaseStrategy, HttpResponseProtocol
 
 
 class BaseAuth:
@@ -31,8 +32,10 @@ class BaseAuth:
     REQUIRES_EMAIL_VALIDATION = False
     SEND_USER_AGENT = True
 
-    def __init__(self, strategy, redirect_uri=None) -> None:
-        self.strategy = strategy
+    def __init__(
+        self, strategy: BaseStrategy | None = None, redirect_uri: str | None = None
+    ) -> None:
+        self.strategy = strategy if strategy is not None else REGISTRY.default_strategy
         self.redirect_uri = redirect_uri
         self.data = self.strategy.request_data()
         self.redirect_uri = self.strategy.absolute_uri(self.redirect_uri)
@@ -115,7 +118,7 @@ def pipeline(
     def disconnect(self, *args, **kwargs) -> dict:
         pipeline = self.strategy.get_disconnect_pipeline(self)
         kwargs["name"] = self.name
-        kwargs["user_storage"] = self.strategy.storage.user
+        kwargs["user_storage"] = cast("type[BaseStorage]", self.strategy.storage).user
         return self.run_pipeline(pipeline, *args, **kwargs)
 
     def run_pipeline(

social_core/backends/discourse.py

@@ -2,13 +2,17 @@
 import time
 from base64 import urlsafe_b64decode, urlsafe_b64encode
 from hashlib import sha256
+from typing import TYPE_CHECKING, cast
 from urllib.parse import urlencode
 
 from social_core.exceptions import AuthException, AuthTokenError
 from social_core.utils import parse_qs
 
 from .base import BaseAuth
 
+if TYPE_CHECKING:
+    from social_core.storage import BaseStorage
+
 
 class DiscourseAuth(BaseAuth):
     name = "discourse"
@@ -51,13 +55,17 @@ def get_user_details(self, response):
         }
 
     def add_nonce(self, nonce) -> None:
-        self.strategy.storage.nonce.use(self.setting("SERVER_URL"), time.time(), nonce)
+        cast("type[BaseStorage]", self.strategy.storage).nonce.use(
+            self.setting("SERVER_URL"), time.time(), nonce
+        )
 
     def get_nonce(self, nonce):
-        return self.strategy.storage.nonce.get(self.setting("SERVER_URL"), nonce)
+        return cast("type[BaseStorage]", self.strategy.storage).nonce.get(
+            self.setting("SERVER_URL"), nonce
+        )
 
     def delete_nonce(self, nonce) -> None:
-        self.strategy.storage.nonce.delete(nonce)
+        cast("type[BaseStorage]", self.strategy.storage).nonce.delete(nonce)
 
     def auth_complete(self, *args, **kwargs):
         """

social_core/backends/open_id_connect.py

@@ -4,9 +4,11 @@
 import datetime
 import json
 from calendar import timegm
-from typing import TYPE_CHECKING, Any, Literal
+from typing import TYPE_CHECKING, Any, Literal, cast
 
 import jwt
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import hashes
 from jwt import (
     ExpiredSignatureError,
     InvalidAudienceError,
@@ -28,8 +30,8 @@
 
     from requests.auth import AuthBase
 
-from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives import hashes
+    from social_core.storage import BaseStorage
+    from social_core.strategy import BaseStrategy
 
 
 class OpenIdConnectAssociation:
@@ -88,7 +90,9 @@ class OpenIdConnectAuth(BaseOAuth2):
     LOGIN_HINT = None
     ACR_VALUES = None
 
-    def __init__(self, strategy, redirect_uri=None) -> None:
+    def __init__(
+        self, strategy: BaseStrategy | None = None, redirect_uri: str | None = None
+    ) -> None:
         super().__init__(strategy, redirect_uri=redirect_uri)
         self.id_token = None
 
@@ -231,19 +235,21 @@ def get_and_store_nonce(self, url, state):
         nonce = self.strategy.random_string(64)
         # Store the nonce
         association = OpenIdConnectAssociation(nonce, assoc_type=state)
-        self.strategy.storage.association.store(url, association)
+        cast("type[BaseStorage]", self.strategy.storage).association.store(
+            url, association
+        )
         return nonce
 
     def get_nonce(self, nonce):
         try:
-            return self.strategy.storage.association.get(
+            return cast("type[BaseStorage]", self.strategy.storage).association.get(
                 server_url=self.authorization_url(), handle=nonce
             )[0]
         except IndexError:
             pass
 
     def remove_nonce(self, nonce_id) -> None:
-        self.strategy.storage.association.remove([nonce_id])
+        cast("type[BaseStorage]", self.strategy.storage).association.remove([nonce_id])
 
     def validate_claims(self, id_token) -> None:
         utc_timestamp = timegm(datetime.datetime.now(datetime.timezone.utc).timetuple())

social_core/backends/vk.py

@@ -207,7 +207,6 @@ def auth_complete(self, *args, **kwargs):
         response = {self.id_key(): user_id}
         response.update(json.loads(request["api_result"])["response"][0])
         return self.strategy.authenticate(
-            *args,
             auth=self,
             backend=self,
             request=request,

social_core/exceptions.py

@@ -22,6 +22,13 @@ def __str__(self) -> str:
         return f"Strategy {self.strategy_name} does not support {self.feature_name}"
 
 
+class DefaultStrategyMissingError(SocialAuthBaseException):
+    """Default strategy is not configured."""
+
+    def __str__(self) -> str:
+        return "Default strategy is not configured"
+
+
 class StrategyMissingBackendError(SocialAuthBaseException):
     """Strategy storage backend is not configured."""
 

social_core/pipeline/social_auth.py

@@ -1,12 +1,12 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from social_core.exceptions import AuthAlreadyAssociated, AuthException, AuthForbidden
 
 if TYPE_CHECKING:
     from social_core.backends.base import BaseAuth
-    from social_core.storage import UserProtocol
+    from social_core.storage import BaseStorage, UserProtocol
 
 
 def social_details(backend: BaseAuth, details, response, *args, **kwargs):
@@ -26,7 +26,9 @@ def social_user(
     backend: BaseAuth, uid, user: UserProtocol | None = None, *args, **kwargs
 ):
     provider = backend.name
-    social = backend.strategy.storage.user.get_social_auth(provider, uid)
+    social = cast("type[BaseStorage]", backend.strategy.storage).user.get_social_auth(
+        provider, uid
+    )
     if social:
         if user and social.user != user:
             raise AuthAlreadyAssociated(backend)
@@ -50,11 +52,13 @@ def associate_user(
 ):
     if user and not social:
         try:
-            social = backend.strategy.storage.user.create_social_auth(
-                user, uid, backend.name
-            )
+            social = cast(
+                "type[BaseStorage]", backend.strategy.storage
+            ).user.create_social_auth(user, uid, backend.name)
         except Exception as err:
-            if not backend.strategy.storage.is_integrity_error(err):
+            if not cast(
+                "type[BaseStorage]", backend.strategy.storage
+            ).is_integrity_error(err):
                 raise
             # Protect for possible race condition, those bastard with FTL
             # clicking capabilities, check issue #131:
@@ -91,7 +95,11 @@ def associate_by_email(
         # Try to associate accounts registered with the same email address,
         # only if it's a single object. AuthException is raised if multiple
         # objects are returned.
-        users = list(backend.strategy.storage.user.get_users_by_email(email))
+        users = list(
+            cast("type[BaseStorage]", backend.strategy.storage).user.get_users_by_email(
+                email
+            )
+        )
         if len(users) == 0:
             return None
         if len(users) > 1:
@@ -111,9 +119,9 @@ def load_extra_data(
     *args,
     **kwargs,
 ) -> None:
-    social = kwargs.get("social") or backend.strategy.storage.user.get_social_auth(
-        backend.name, uid
-    )
+    social = kwargs.get("social") or cast(
+        "type[BaseStorage]", backend.strategy.storage
+    ).user.get_social_auth(backend.name, uid)
     if social:
         extra_data = backend.extra_data(user, uid, response, details, kwargs)
         social.set_extra_data(extra_data)

social_core/pipeline/user.py

@@ -10,7 +10,7 @@
 
 if TYPE_CHECKING:
     from social_core.backends.base import BaseAuth
-    from social_core.storage import UserProtocol
+    from social_core.storage import BaseStorage, UserProtocol
     from social_core.strategy import BaseStrategy
 
 USER_FIELDS = ["username", "email"]
@@ -173,4 +173,4 @@ def user_details(
         setattr(user, name, value)
 
     if changed:
-        strategy.storage.user.changed(user)
+        cast("type[BaseStorage]", strategy.storage).user.changed(user)

social_core/pipeline/utils.py

@@ -1,14 +1,14 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from social_core.exceptions import (
     StrategyMissingBackendError,
 )
 
 if TYPE_CHECKING:
     from social_core.backends.base import BaseAuth
-    from social_core.storage import PartialMixin, UserProtocol
+    from social_core.storage import BaseStorage, PartialMixin, UserProtocol
     from social_core.strategy import BaseStrategy
 
 SERIALIZABLE_TYPES = (dict, list, tuple, set, bool, type(None), int, str, bytes)
@@ -84,10 +84,14 @@ def partial_load(strategy: BaseStrategy, token: str) -> PartialMixin | None:
         social = kwargs.get("social")
 
         if isinstance(social, dict):
-            kwargs["social"] = strategy.storage.user.get_social_auth(**social)  # type: ignore[missing-argument]
+            kwargs["social"] = cast(
+                "type[BaseStorage]", strategy.storage
+            ).user.get_social_auth(**social)  # type: ignore[missing-argument]
 
         if user:
-            kwargs["user"] = strategy.storage.user.get_user(user)
+            kwargs["user"] = cast("type[BaseStorage]", strategy.storage).user.get_user(
+                user
+            )
 
         partial.args = [strategy.from_session_value(val) for val in args]
         partial.kwargs = {

social_core/registry.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from .exceptions import DefaultStrategyMissingError
+
+if TYPE_CHECKING:
+    from .strategy import BaseStrategy
+
+
+class Registry:
+    def __init__(self) -> None:
+        self._default_strategy: BaseStrategy | None = None
+
+    @property
+    def default_strategy(self) -> BaseStrategy:
+        if self._default_strategy is None:
+            raise DefaultStrategyMissingError
+        return self._default_strategy
+
+    @default_strategy.setter
+    def default_strategy(self, strategy: BaseStrategy) -> None:
+        self._default_strategy = strategy
+
+
+REGISTRY = Registry()