chore: improve type annotations

This is needed for proper type annotations in #1491, see also #1490.

Author: nijel

social_core/actions.py

@@ -12,8 +12,10 @@
 )
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from .backends.base import BaseAuth
-    from .storage import UserProtocol
+    from .storage import BaseStorage, UserProtocol
     from .strategy import HttpResponseProtocol
 
 
@@ -22,7 +24,9 @@ def do_auth(backend: BaseAuth, redirect_name: str = "next") -> HttpResponseProto
     data = backend.strategy.request_data(merge=False)
 
     # Save extra data into session.
-    for field_name in backend.setting("FIELDS_STORED_IN_SESSION", []):
+    for field_name in cast(
+        "list[str]", backend.setting("FIELDS_STORED_IN_SESSION", [])
+    ):
         if field_name in data:
             backend.strategy.session_set(field_name, data[field_name])
         else:
@@ -33,7 +37,7 @@ def do_auth(backend: BaseAuth, redirect_name: str = "next") -> HttpResponseProto
         redirect_uri = data[redirect_name]
         if backend.setting("SANITIZE_REDIRECTS", True):
             allowed_hosts = [
-                *backend.setting("ALLOWED_REDIRECT_HOSTS", []),
+                *cast("list[str]", backend.setting("ALLOWED_REDIRECT_HOSTS", [])),
                 backend.strategy.request_host(),
             ]
             redirect_uri = sanitize_redirect(allowed_hosts, redirect_uri)
@@ -43,9 +47,9 @@ def do_auth(backend: BaseAuth, redirect_name: str = "next") -> HttpResponseProto
     return backend.start()
 
 
-def do_complete(
+def do_complete(  # noqa: C901,PLR0912
     backend: BaseAuth,
-    login,
+    login: Callable,
     user: UserProtocol | None = None,
     redirect_name: str = "next",
     *args,
@@ -54,15 +58,19 @@ def do_complete(
     data = backend.strategy.request_data()
 
     is_authenticated = user_is_authenticated(user)
-    user = user if is_authenticated else None
+    authenticated_user: UserProtocol | HttpResponseProtocol | None = (
+        user if is_authenticated else None
+    )
 
     partial = partial_pipeline_data(backend, user, *args, **kwargs)
     if partial:
-        user = backend.continue_pipeline(partial)
+        authenticated_user = backend.continue_pipeline(partial)
         # clean partial data after usage
         backend.strategy.clean_partial_pipeline(partial.token)
     else:
-        user = backend.complete(*args, user=user, redirect_name=redirect_name, **kwargs)
+        authenticated_user = backend.complete(
+            *args, user=authenticated_user, redirect_name=redirect_name, **kwargs
+        )
 
     # pop redirect value before the session is trashed on login(), but after
     # the pipeline so that the pipeline can change the redirect if needed
@@ -72,12 +80,15 @@ 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()
-    if user and not isinstance(user, user_model):
-        return cast("HttpResponseProtocol", user)
+    user_model = cast("type[BaseStorage]", backend.strategy.storage).user.user_model()
+    if authenticated_user and not isinstance(authenticated_user, user_model):
+        return cast("HttpResponseProtocol", authenticated_user)
+
+    authenticated_user = cast("UserProtocol | None", authenticated_user)
+    url: str | None
 
     if is_authenticated:
-        if not user:
+        if not authenticated_user:
             url = setting_url(backend, redirect_value, "LOGIN_REDIRECT_URL")
         else:
             url = setting_url(
@@ -86,17 +97,17 @@ def do_complete(
                 "NEW_ASSOCIATION_REDIRECT_URL",
                 "LOGIN_REDIRECT_URL",
             )
-    elif user:
+    elif authenticated_user:
         # check if inactive users are allowed to login
         bypass_inactivation = backend.strategy.setting(
             "ALLOW_INACTIVE_USERS_LOGIN", False
         )
-        if bypass_inactivation or user_is_active(user):
+        if bypass_inactivation or user_is_active(authenticated_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  # type: ignore[union-attr]
-            login(backend, user, social_user)
+            is_new = getattr(authenticated_user, "is_new", False)
+            social_user = authenticated_user.social_user  # type: ignore[union-attr]
+            login(backend, authenticated_user, social_user)
             # store last login backend name in session
             backend.strategy.session_set(
                 "social_auth_last_login_backend", social_user.provider
@@ -114,28 +125,31 @@ def do_complete(
         else:
             if backend.setting("INACTIVE_USER_LOGIN", False):
                 # This attribute is set in BaseAuth.pipeline()
-                social_user = user.social_user  # type: ignore[union-attr]
-                login(backend, user, social_user)
+                social_user = authenticated_user.social_user  # type: ignore[union-attr]
+                login(backend, authenticated_user, social_user)
             url = setting_url(
                 backend, "INACTIVE_USER_URL", "LOGIN_ERROR_URL", "LOGIN_URL"
             )
     else:
         url = setting_url(backend, "LOGIN_ERROR_URL", "LOGIN_URL")
 
-    assert url, "By this point URL has to have been set"
+    if not url:
+        raise ValueError("By this point URL has to have been set")
 
     if redirect_value and redirect_value != url:
         redirect_value = quote(redirect_value)
         url += ("&" if "?" in url else "?") + f"{redirect_name}={redirect_value}"
 
     if backend.setting("SANITIZE_REDIRECTS", True):
         allowed_hosts = [
-            *backend.setting("ALLOWED_REDIRECT_HOSTS", []),
+            *cast("list[str]", backend.setting("ALLOWED_REDIRECT_HOSTS", [])),
             backend.strategy.request_host(),
         ]
         url = sanitize_redirect(allowed_hosts, url) or backend.setting(
             "LOGIN_REDIRECT_URL"
         )
+        if url is None:
+            raise ValueError("Disallowed URL")
     return backend.strategy.redirect(url)
 
 
@@ -160,20 +174,22 @@ def do_disconnect(
         )
 
     if isinstance(response, dict):
-        url = backend.strategy.absolute_uri(
+        url: str | None = backend.strategy.absolute_uri(
             backend.strategy.request_data().get(redirect_name, "")
             or backend.setting("DISCONNECT_REDIRECT_URL")
             or backend.setting("LOGIN_REDIRECT_URL")
         )
         if backend.setting("SANITIZE_REDIRECTS", True):
             allowed_hosts = [
-                *backend.setting("ALLOWED_REDIRECT_HOSTS", []),
+                *cast("list[str]", backend.setting("ALLOWED_REDIRECT_HOSTS", [])),
                 backend.strategy.request_host(),
             ]
             url = (
                 sanitize_redirect(allowed_hosts, url)
                 or backend.setting("DISCONNECT_REDIRECT_URL")
                 or backend.setting("LOGIN_REDIRECT_URL")
             )
+        if not url:
+            raise ValueError("Disallowed URL")
         response = backend.strategy.redirect(url)
     return response

social_core/backends/apple.py

@@ -25,7 +25,7 @@
 
 import json
 import time
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 import jwt
 from jwt.algorithms import RSAAlgorithm  # ty: ignore[possibly-missing-import]
@@ -71,12 +71,12 @@ def auth_params(self, *args, **kwargs):
             params["response_mode"] = "form_post"
         return params
 
-    def get_private_key(self):
+    def get_private_key(self) -> str:
         """
         Return contents of the private key file. Override this method to provide
         secret key from another source if needed.
         """
-        return self.setting("SECRET")
+        return cast("str", self.setting("SECRET"))
 
     def generate_client_secret(self):
         now = int(time.time())

social_core/backends/azuread_b2c.py

@@ -30,7 +30,7 @@
 from __future__ import annotations
 
 import json
-from typing import TYPE_CHECKING, Any, Literal
+from typing import TYPE_CHECKING, Any, Literal, cast
 
 from cryptography.hazmat.primitives import serialization
 from jwt import DecodeError, ExpiredSignatureError, get_unverified_header
@@ -198,7 +198,7 @@ def user_data(self, access_token: str, *args, **kwargs) -> dict[str, Any] | None
                 key=key,
                 algorithms=["RS256"],
                 audience=self.setting("KEY"),
-                leeway=self.setting("JWT_LEEWAY", default=0),
+                leeway=cast("int", self.setting("JWT_LEEWAY", default=0)),
             )
         except (DecodeError, ExpiredSignatureError) as error:
             raise AuthTokenError(self, error)

social_core/backends/azuread_tenant.py

@@ -1,5 +1,5 @@
 import base64
-from typing import Any
+from typing import Any, cast
 
 from cryptography.hazmat.backends import default_backend
 from cryptography.x509 import load_der_x509_certificate
@@ -51,8 +51,8 @@ class AzureADTenantOAuth2(AzureADOAuth2):
     JWKS_URL = "{base_url}/discovery/keys{appid}"
 
     @property
-    def tenant_id(self):
-        return self.setting("TENANT_ID", "common")
+    def tenant_id(self) -> str:
+        return cast("str", self.setting("TENANT_ID", "common"))
 
     def openid_configuration_url(self):
         return self.OPENID_CONFIGURATION_URL.format(

social_core/backends/base.py

@@ -8,7 +8,7 @@
 
 from social_core.exceptions import AuthConnectionError, AuthUnknownError
 from social_core.registry import REGISTRY
-from social_core.storage import UserProtocol
+from social_core.storage import BaseStorage, UserProtocol
 from social_core.utils import module_member, parse_qs, social_logger, user_agent
 
 if TYPE_CHECKING:
@@ -17,8 +17,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, PartialMixin, UserProtocol
+    from social_core.strategy import BaseStrategy, HttpResponseProtocol
 
 
 class BaseAuth:
@@ -33,8 +33,9 @@ class BaseAuth:
     REQUIRES_EMAIL_VALIDATION = False
     SEND_USER_AGENT = True
 
-    def __init__(self, strategy=None, redirect_uri: str | None = None) -> None:
-        # TODO: temporary type override
+    def __init__(
+        self, strategy: BaseStrategy | None = None, redirect_uri: str | None = None
+    ) -> None:
         self.strategy: Any = (
             strategy if strategy is not None else REGISTRY.default_strategy
         )
@@ -57,7 +58,7 @@ def start(self) -> HttpResponseProtocol:
             return self.strategy.redirect(self.auth_url())
         return self.strategy.html(self.auth_html())
 
-    def complete(self, *args, **kwargs) -> UserProtocol | None:
+    def complete(self, *args, **kwargs) -> HttpResponseProtocol | UserProtocol | None:
         return self.auth_complete(*args, **kwargs)
 
     def auth_url(self) -> str:
@@ -68,7 +69,9 @@ def auth_html(self) -> str:
         """Must return login HTML content returned by provider"""
         return "Implement in subclass"
 
-    def auth_complete(self, *args, **kwargs) -> UserProtocol | None:
+    def auth_complete(
+        self, *args, **kwargs
+    ) -> HttpResponseProtocol | UserProtocol | None:
         """Completes login process, must return user instance"""
         raise NotImplementedError("Implement in subclass")
 
@@ -120,7 +123,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(
@@ -194,8 +197,14 @@ def extra_data(
     def auth_allowed(self, response, details):
         """Return True if the user should be allowed to authenticate, by
         default check if email is whitelisted (if there's a whitelist)"""
-        emails = [email.lower() for email in self.setting("WHITELISTED_EMAILS", [])]
-        domains = [domain.lower() for domain in self.setting("WHITELISTED_DOMAINS", [])]
+        emails = [
+            email.lower()
+            for email in cast("list[str]", self.setting("WHITELISTED_EMAILS", []))
+        ]
+        domains = [
+            domain.lower()
+            for domain in cast("list[str]", self.setting("WHITELISTED_DOMAINS", []))
+        ]
         email = details.get("email")
         allowed = True
         if email and (emails or domains):
@@ -249,7 +258,9 @@ def get_user(self, user_id):
         """
         return self.strategy.get_user(user_id)
 
-    def continue_pipeline(self, partial):
+    def continue_pipeline(
+        self, partial: PartialMixin
+    ) -> UserProtocol | HttpResponseProtocol | None:
         """Continue previous halted pipeline"""
         return self.strategy.authenticate(
             self, *partial.args, pipeline_index=partial.next_step, **partial.kwargs
@@ -336,7 +347,7 @@ def get_key_and_secret(self) -> tuple[str, str]:
         """Return tuple with Consumer Key and Consumer Secret for current
         service provider. Must return (key, secret), order *must* be respected.
         """
-        return self.setting("KEY"), self.setting("SECRET")
+        return cast("str", self.setting("KEY")), cast("str", self.setting("SECRET"))
 
     def get_key_and_secret_basic_auth(self) -> bytes:
         """Generate HTTP Basic Authentication header value from KEY and SECRET.

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):
         """