diff --git a/authentik/admin/api/metrics.py b/authentik/admin/api/metrics.py index 333ea79f8..42909017d 100644 --- a/authentik/admin/api/metrics.py +++ b/authentik/admin/api/metrics.py @@ -2,7 +2,7 @@ import time from collections import Counter from datetime import timedelta -from typing import Dict, List +from typing import List from django.db.models import Count, ExpressionWrapper, F, Model from django.db.models.fields import DurationField @@ -19,7 +19,7 @@ from rest_framework.viewsets import ViewSet from authentik.events.models import Event, EventAction -def get_events_per_1h(**filter_kwargs) -> List[Dict[str, int]]: +def get_events_per_1h(**filter_kwargs) -> List[dict[str, int]]: """Get event count by hour in the last day, fill with zeros""" date_from = now() - timedelta(days=1) result = ( diff --git a/authentik/admin/views/outposts.py b/authentik/admin/views/outposts.py index fb9eee561..5cc7bd8eb 100644 --- a/authentik/admin/views/outposts.py +++ b/authentik/admin/views/outposts.py @@ -1,6 +1,6 @@ """authentik Outpost administration""" from dataclasses import asdict -from typing import Any, Dict +from typing import Any from django.contrib.auth.mixins import LoginRequiredMixin from django.contrib.auth.mixins import ( @@ -33,7 +33,7 @@ class OutpostCreateView( template_name = "generic/create.html" success_message = _("Successfully created Outpost") - def get_initial(self) -> Dict[str, Any]: + def get_initial(self) -> dict[str, Any]: return { "_config": asdict( OutpostConfig(authentik_host=self.request.build_absolute_uri("/")) diff --git a/authentik/admin/views/policies.py b/authentik/admin/views/policies.py index 4c7301303..3a999f3cc 100644 --- a/authentik/admin/views/policies.py +++ b/authentik/admin/views/policies.py @@ -1,5 +1,5 @@ """authentik Policy administration""" -from typing import Any, Dict +from typing import Any from django.contrib.auth.mixins import LoginRequiredMixin from django.contrib.auth.mixins import ( @@ -102,7 +102,7 @@ class PolicyTestView(LoginRequiredMixin, DetailView, PermissionRequiredMixin, Fo Policy.objects.filter(pk=self.kwargs.get("pk")).select_subclasses().first() ) - def get_context_data(self, **kwargs: Any) -> Dict[str, Any]: + def get_context_data(self, **kwargs: Any) -> dict[str, Any]: kwargs["policy"] = self.get_object() return super().get_context_data(**kwargs) diff --git a/authentik/admin/views/tasks.py b/authentik/admin/views/tasks.py index 40dfa74c1..fdc69d5f4 100644 --- a/authentik/admin/views/tasks.py +++ b/authentik/admin/views/tasks.py @@ -1,5 +1,5 @@ """authentik Tasks List""" -from typing import Any, Dict +from typing import Any from django.views.generic.base import TemplateView @@ -12,7 +12,7 @@ class TaskListView(AdminRequiredMixin, TemplateView): template_name = "administration/task/list.html" - def get_context_data(self, **kwargs: Any) -> Dict[str, Any]: + def get_context_data(self, **kwargs: Any) -> dict[str, Any]: kwargs = super().get_context_data(**kwargs) kwargs["object_list"] = sorted( TaskInfo.all().values(), key=lambda x: x.task_name diff --git a/authentik/admin/views/utils.py b/authentik/admin/views/utils.py index 1eda77d6c..8ec3bb605 100644 --- a/authentik/admin/views/utils.py +++ b/authentik/admin/views/utils.py @@ -1,5 +1,5 @@ """authentik admin util views""" -from typing import Any, Dict, List, Optional +from typing import Any, List, Optional from urllib.parse import urlparse from django.contrib import messages @@ -69,7 +69,7 @@ class InheritanceCreateView(CreateAssignPermView): raise Http404 from exc return model().form - def get_context_data(self, **kwargs: Any) -> Dict[str, Any]: + def get_context_data(self, **kwargs: Any) -> dict[str, Any]: kwargs = super().get_context_data(**kwargs) form_cls = self.get_form_class() if hasattr(form_cls, "template_name"): @@ -80,7 +80,7 @@ class InheritanceCreateView(CreateAssignPermView): class InheritanceUpdateView(UpdateView): """UpdateView for objects using InheritanceManager""" - def get_context_data(self, **kwargs: Any) -> Dict[str, Any]: + def get_context_data(self, **kwargs: Any) -> dict[str, Any]: kwargs = super().get_context_data(**kwargs) form_cls = self.get_form_class() if hasattr(form_cls, "template_name"): diff --git a/authentik/core/models.py b/authentik/core/models.py index eb0ee27d4..ac88f594b 100644 --- a/authentik/core/models.py +++ b/authentik/core/models.py @@ -1,7 +1,7 @@ """authentik core models""" from datetime import timedelta from hashlib import sha256 -from typing import Any, Dict, Optional, Type +from typing import Any, Optional, Type from uuid import uuid4 from django.conf import settings @@ -96,7 +96,7 @@ class User(GuardianUserMixin, AbstractUser): objects = UserManager() - def group_attributes(self) -> Dict[str, Any]: + def group_attributes(self) -> dict[str, Any]: """Get a dictionary containing the attributes from all groups the user belongs to, including the users attributes""" final_attributes = {} diff --git a/authentik/core/views/user.py b/authentik/core/views/user.py index 9112e1ea2..140c6e6da 100644 --- a/authentik/core/views/user.py +++ b/authentik/core/views/user.py @@ -1,5 +1,5 @@ """authentik core user views""" -from typing import Any, Dict +from typing import Any from django.contrib.auth.mixins import LoginRequiredMixin from django.contrib.auth.mixins import ( @@ -45,7 +45,7 @@ class UserDetailsView(SuccessMessageMixin, LoginRequiredMixin, UpdateView): def get_object(self): return self.request.user - def get_context_data(self, **kwargs: Any) -> Dict[str, Any]: + def get_context_data(self, **kwargs: Any) -> dict[str, Any]: kwargs = super().get_context_data(**kwargs) unenrollment_flow = Flow.with_policy( self.request, designation=FlowDesignation.UNRENOLLMENT diff --git a/authentik/events/monitored_tasks.py b/authentik/events/monitored_tasks.py index 8842d1fc5..dd42a4799 100644 --- a/authentik/events/monitored_tasks.py +++ b/authentik/events/monitored_tasks.py @@ -3,7 +3,7 @@ from dataclasses import dataclass, field from datetime import datetime from enum import Enum from traceback import format_tb -from typing import Any, Dict, List, Optional +from typing import Any, List, Optional from celery import Task from django.core.cache import cache @@ -50,7 +50,7 @@ class TaskInfo: task_call_module: str task_call_func: str task_call_args: List[Any] = field(default_factory=list) - task_call_kwargs: Dict[str, Any] = field(default_factory=dict) + task_call_kwargs: dict[str, Any] = field(default_factory=dict) task_description: Optional[str] = field(default=None) @@ -60,7 +60,7 @@ class TaskInfo: return self.task_name.split("_") @staticmethod - def all() -> Dict[str, "TaskInfo"]: + def all() -> dict[str, "TaskInfo"]: """Get all TaskInfo objects""" return cache.get_many(cache.keys("task_*")) @@ -109,7 +109,7 @@ class MonitoredTask(Task): # pylint: disable=too-many-arguments def after_return( - self, status, retval, task_id, args: List[Any], kwargs: Dict[str, Any], einfo + self, status, retval, task_id, args: List[Any], kwargs: dict[str, Any], einfo ): if not self._result.uid: self._result.uid = self._uid diff --git a/authentik/events/signals.py b/authentik/events/signals.py index d2ef3b6cf..c0f7124aa 100644 --- a/authentik/events/signals.py +++ b/authentik/events/signals.py @@ -1,6 +1,6 @@ """authentik events signal listener""" from threading import Thread -from typing import Any, Dict, Optional +from typing import Any, Optional from django.contrib.auth.signals import ( user_logged_in, @@ -27,7 +27,7 @@ class EventNewThread(Thread): action: str request: HttpRequest - kwargs: Dict[str, Any] + kwargs: dict[str, Any] user: Optional[User] = None def __init__( @@ -69,7 +69,7 @@ def on_user_logged_out(sender, request: HttpRequest, user: User, **_): @receiver(user_write) # pylint: disable=unused-argument def on_user_write( - sender, request: HttpRequest, user: User, data: Dict[str, Any], **kwargs + sender, request: HttpRequest, user: User, data: dict[str, Any], **kwargs ): """Log User write""" thread = EventNewThread(EventAction.USER_WRITE, request, **data) @@ -81,7 +81,7 @@ def on_user_write( @receiver(user_login_failed) # pylint: disable=unused-argument def on_user_login_failed( - sender, credentials: Dict[str, str], request: HttpRequest, **_ + sender, credentials: dict[str, str], request: HttpRequest, **_ ): """Failed Login""" thread = EventNewThread(EventAction.LOGIN_FAILED, request, **credentials) diff --git a/authentik/events/utils.py b/authentik/events/utils.py index 3305303b8..29974c693 100644 --- a/authentik/events/utils.py +++ b/authentik/events/utils.py @@ -1,7 +1,7 @@ """event utilities""" import re from dataclasses import asdict, is_dataclass -from typing import Any, Dict, Optional +from typing import Any, Optional from uuid import UUID from django.contrib.auth.models import AnonymousUser @@ -20,7 +20,7 @@ from authentik.policies.types import PolicyRequest ALLOWED_SPECIAL_KEYS = re.compile("passing", flags=re.I) -def cleanse_dict(source: Dict[Any, Any]) -> Dict[Any, Any]: +def cleanse_dict(source: dict[Any, Any]) -> dict[Any, Any]: """Cleanse a dictionary, recursively""" final_dict = {} for key, value in source.items(): @@ -38,7 +38,7 @@ def cleanse_dict(source: Dict[Any, Any]) -> Dict[Any, Any]: return final_dict -def model_to_dict(model: Model) -> Dict[str, Any]: +def model_to_dict(model: Model) -> dict[str, Any]: """Convert model to dict""" name = str(model) if hasattr(model, "name"): @@ -51,7 +51,7 @@ def model_to_dict(model: Model) -> Dict[str, Any]: } -def get_user(user: User, original_user: Optional[User] = None) -> Dict[str, Any]: +def get_user(user: User, original_user: Optional[User] = None) -> dict[str, Any]: """Convert user object to dictionary, optionally including the original user""" if isinstance(user, AnonymousUser): user = get_anonymous_user() @@ -67,7 +67,7 @@ def get_user(user: User, original_user: Optional[User] = None) -> Dict[str, Any] return user_data -def sanitize_dict(source: Dict[Any, Any]) -> Dict[Any, Any]: +def sanitize_dict(source: dict[Any, Any]) -> dict[Any, Any]: """clean source of all Models that would interfere with the JSONField. Models are replaced with a dictionary of { app: str, diff --git a/authentik/flows/planner.py b/authentik/flows/planner.py index 0b9318ccf..e54199d8d 100644 --- a/authentik/flows/planner.py +++ b/authentik/flows/planner.py @@ -1,6 +1,6 @@ """Flows Planner""" from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional +from typing import Any, List, Optional from django.core.cache import cache from django.http import HttpRequest @@ -39,7 +39,7 @@ class FlowPlan: flow_pk: str stages: List[Stage] = field(default_factory=list) - context: Dict[str, Any] = field(default_factory=dict) + context: dict[str, Any] = field(default_factory=dict) markers: List[StageMarker] = field(default_factory=list) def append(self, stage: Stage, marker: Optional[StageMarker] = None): @@ -96,7 +96,7 @@ class FlowPlanner: self._logger = get_logger().bind(flow=flow) def plan( - self, request: HttpRequest, default_context: Optional[Dict[str, Any]] = None + self, request: HttpRequest, default_context: Optional[dict[str, Any]] = None ) -> FlowPlan: """Check each of the flows' policies, check policies for each stage with PolicyBinding and return ordered list""" @@ -149,7 +149,7 @@ class FlowPlanner: self, user: User, request: HttpRequest, - default_context: Optional[Dict[str, Any]], + default_context: Optional[dict[str, Any]], ) -> FlowPlan: """Build flow plan by checking each stage in their respective order and checking the applied policies""" diff --git a/authentik/flows/stage.py b/authentik/flows/stage.py index c4e41e1b1..f6014e186 100644 --- a/authentik/flows/stage.py +++ b/authentik/flows/stage.py @@ -1,6 +1,6 @@ """authentik stage Base view""" from collections import namedtuple -from typing import Any, Dict +from typing import Any from django.http import HttpRequest from django.utils.translation import gettext_lazy as _ @@ -26,7 +26,7 @@ class StageView(TemplateView): def __init__(self, executor: FlowExecutorView): self.executor = executor - def get_context_data(self, **kwargs: Any) -> Dict[str, Any]: + def get_context_data(self, **kwargs: Any) -> dict[str, Any]: kwargs["title"] = self.executor.flow.title # Either show the matched User object or show what the user entered, # based on what the earlier stage (mostly IdentificationStage) set. diff --git a/authentik/flows/transfer/common.py b/authentik/flows/transfer/common.py index 7309e4d30..8ff94511a 100644 --- a/authentik/flows/transfer/common.py +++ b/authentik/flows/transfer/common.py @@ -1,6 +1,6 @@ """transfer common classes""" from dataclasses import asdict, dataclass, field, is_dataclass -from typing import Any, Dict, List +from typing import Any, List from uuid import UUID from django.core.serializers.json import DjangoJSONEncoder @@ -9,7 +9,7 @@ from authentik.lib.models import SerializerModel from authentik.lib.sentry import SentryIgnoredException -def get_attrs(obj: SerializerModel) -> Dict[str, Any]: +def get_attrs(obj: SerializerModel) -> dict[str, Any]: """Get object's attributes via their serializer, and covert it to a normal dict""" data = dict(obj.serializer(obj).data) to_remove = ( @@ -33,9 +33,9 @@ def get_attrs(obj: SerializerModel) -> Dict[str, Any]: class FlowBundleEntry: """Single entry of a bundle""" - identifiers: Dict[str, Any] + identifiers: dict[str, Any] model: str - attrs: Dict[str, Any] + attrs: dict[str, Any] @staticmethod def from_model( diff --git a/authentik/flows/transfer/importer.py b/authentik/flows/transfer/importer.py index 2b804f394..2839668a8 100644 --- a/authentik/flows/transfer/importer.py +++ b/authentik/flows/transfer/importer.py @@ -2,7 +2,7 @@ from contextlib import contextmanager from copy import deepcopy from json import loads -from typing import Any, Dict, Type +from typing import Any, Type from dacite import from_dict from dacite.exceptions import DaciteError @@ -42,7 +42,7 @@ class FlowImporter: __import: FlowBundle - __pk_map: Dict[Any, Model] + __pk_map: dict[Any, Model] logger: BoundLogger @@ -55,7 +55,7 @@ class FlowImporter: except DaciteError as exc: raise EntryInvalidError from exc - def __update_pks_for_attrs(self, attrs: Dict[str, Any]) -> Dict[str, Any]: + def __update_pks_for_attrs(self, attrs: dict[str, Any]) -> dict[str, Any]: """Replace any value if it is a known primary key of an other object""" def updater(value) -> Any: @@ -75,7 +75,7 @@ class FlowImporter: attrs[key] = updater(value) return attrs - def __query_from_identifier(self, attrs: Dict[str, Any]) -> Q: + def __query_from_identifier(self, attrs: dict[str, Any]) -> Q: """Generate an or'd query from all identifiers in an entry""" # Since identifiers can also be pk-references to other objects (see FlowStageBinding) # we have to ensure those references are also replaced diff --git a/authentik/flows/views.py b/authentik/flows/views.py index ab98ba850..a2af680c7 100644 --- a/authentik/flows/views.py +++ b/authentik/flows/views.py @@ -1,6 +1,6 @@ """authentik multi-stage authentication engine""" from traceback import format_tb -from typing import Any, Dict, Optional +from typing import Any, Optional from django.contrib.auth.mixins import LoginRequiredMixin from django.http import ( @@ -224,8 +224,8 @@ class FlowErrorResponse(TemplateResponse): self.error = error def resolve_context( - self, context: Optional[Dict[str, Any]] - ) -> Optional[Dict[str, Any]]: + self, context: Optional[dict[str, Any]] + ) -> Optional[dict[str, Any]]: if not context: context = {} context["error"] = self.error @@ -243,7 +243,7 @@ class FlowExecutorShellView(TemplateView): template_name = "flows/shell.html" - def get_context_data(self, **kwargs) -> Dict[str, Any]: + def get_context_data(self, **kwargs) -> dict[str, Any]: flow: Flow = get_object_or_404(Flow, slug=self.kwargs.get("flow_slug")) kwargs["background_url"] = flow.background.url kwargs["exec_url"] = reverse( diff --git a/authentik/lib/expression/evaluator.py b/authentik/lib/expression/evaluator.py index b94fe53c1..27db18023 100644 --- a/authentik/lib/expression/evaluator.py +++ b/authentik/lib/expression/evaluator.py @@ -1,7 +1,7 @@ """authentik expression policy evaluator""" import re from textwrap import indent -from typing import Any, Dict, Iterable, Optional +from typing import Any, Iterable, Optional from django.core.exceptions import ValidationError from requests import Session @@ -18,9 +18,9 @@ class BaseEvaluator: """Validate and evaluate python-based expressions""" # Globals that can be used by function - _globals: Dict[str, Any] + _globals: dict[str, Any] # Context passed as locals to exec() - _context: Dict[str, Any] + _context: dict[str, Any] # Filename used for exec _filename: str diff --git a/authentik/lib/utils/http.py b/authentik/lib/utils/http.py index 4039b94c6..35e5a32a6 100644 --- a/authentik/lib/utils/http.py +++ b/authentik/lib/utils/http.py @@ -1,10 +1,10 @@ """http helpers""" -from typing import Any, Dict, Optional +from typing import Any, Optional from django.http import HttpRequest -def _get_client_ip_from_meta(meta: Dict[str, Any]) -> Optional[str]: +def _get_client_ip_from_meta(meta: dict[str, Any]) -> Optional[str]: """Attempt to get the client's IP by checking common HTTP Headers. Returns none if no IP Could be found""" headers = ( diff --git a/authentik/outposts/channels.py b/authentik/outposts/channels.py index 5b013a4e5..7ad5b0dfa 100644 --- a/authentik/outposts/channels.py +++ b/authentik/outposts/channels.py @@ -2,7 +2,7 @@ from dataclasses import asdict, dataclass, field from datetime import datetime from enum import IntEnum -from typing import Any, Dict, Optional +from typing import Any, Optional from channels.exceptions import DenyConnection from dacite import from_dict @@ -34,7 +34,7 @@ class WebsocketMessage: """Complete Websocket Message that is being sent""" instruction: int - args: Dict[str, Any] = field(default_factory=dict) + args: dict[str, Any] = field(default_factory=dict) class OutpostConsumer(AuthJsonConsumer): diff --git a/authentik/outposts/controllers/docker.py b/authentik/outposts/controllers/docker.py index 77c46e431..a4c57b3e5 100644 --- a/authentik/outposts/controllers/docker.py +++ b/authentik/outposts/controllers/docker.py @@ -1,6 +1,6 @@ """Docker controller""" from time import sleep -from typing import Dict, Tuple +from typing import Tuple from django.conf import settings from docker import DockerClient @@ -33,10 +33,10 @@ class DockerController(BaseController): except ServiceConnectionInvalid as exc: raise ControllerException from exc - def _get_labels(self) -> Dict[str, str]: + def _get_labels(self) -> dict[str, str]: return {} - def _get_env(self) -> Dict[str, str]: + def _get_env(self) -> dict[str, str]: return { "AUTHENTIK_HOST": self.outpost.config.authentik_host, "AUTHENTIK_INSECURE": str(self.outpost.config.authentik_host_insecure), diff --git a/authentik/outposts/controllers/k8s/deployment.py b/authentik/outposts/controllers/k8s/deployment.py index 022e9bce3..c01a00881 100644 --- a/authentik/outposts/controllers/k8s/deployment.py +++ b/authentik/outposts/controllers/k8s/deployment.py @@ -1,5 +1,5 @@ """Kubernetes Deployment Reconciler""" -from typing import TYPE_CHECKING, Dict +from typing import TYPE_CHECKING from kubernetes.client import ( AppsV1Api, @@ -53,7 +53,7 @@ class DeploymentReconciler(KubernetesObjectReconciler[V1Deployment]): ): raise NeedsUpdate() - def get_pod_meta(self) -> Dict[str, str]: + def get_pod_meta(self) -> dict[str, str]: """Get common object metadata""" return { "app.kubernetes.io/name": "authentik-outpost", diff --git a/authentik/outposts/controllers/kubernetes.py b/authentik/outposts/controllers/kubernetes.py index f75edf823..b476ca23b 100644 --- a/authentik/outposts/controllers/kubernetes.py +++ b/authentik/outposts/controllers/kubernetes.py @@ -1,6 +1,6 @@ """Kubernetes deployment controller""" from io import StringIO -from typing import Dict, List, Type +from typing import List, Type from kubernetes.client import OpenApiException from kubernetes.client.api_client import ApiClient @@ -18,7 +18,7 @@ from authentik.outposts.models import KubernetesServiceConnection, Outpost class KubernetesController(BaseController): """Manage deployment of outpost in kubernetes""" - reconcilers: Dict[str, Type[KubernetesObjectReconciler]] + reconcilers: dict[str, Type[KubernetesObjectReconciler]] reconcile_order: List[str] client: ApiClient diff --git a/authentik/outposts/models.py b/authentik/outposts/models.py index eb2056dff..694d78b00 100644 --- a/authentik/outposts/models.py +++ b/authentik/outposts/models.py @@ -1,7 +1,7 @@ """Outpost models""" from dataclasses import asdict, dataclass, field from datetime import datetime -from typing import Dict, Iterable, List, Optional, Type, Union +from typing import Iterable, List, Optional, Type, Union from uuid import uuid4 from dacite import from_dict @@ -58,7 +58,7 @@ class OutpostConfig: kubernetes_replicas: int = field(default=1) kubernetes_namespace: str = field(default="default") - kubernetes_ingress_annotations: Dict[str, str] = field(default_factory=dict) + kubernetes_ingress_annotations: dict[str, str] = field(default_factory=dict) kubernetes_ingress_secret_name: str = field(default="authentik-outpost") diff --git a/authentik/policies/http.py b/authentik/policies/http.py index 72a0b5f82..5da2a30ce 100644 --- a/authentik/policies/http.py +++ b/authentik/policies/http.py @@ -1,5 +1,5 @@ """policy http response""" -from typing import Any, Dict, Optional +from typing import Any, Optional from django.http.request import HttpRequest from django.template.response import TemplateResponse @@ -24,8 +24,8 @@ class AccessDeniedResponse(TemplateResponse): self.title = _("Access denied") def resolve_context( - self, context: Optional[Dict[str, Any]] - ) -> Optional[Dict[str, Any]]: + self, context: Optional[dict[str, Any]] + ) -> Optional[dict[str, Any]]: if not context: context = {} context["title"] = self.title diff --git a/authentik/policies/utils.py b/authentik/policies/utils.py index 9643fbcc9..c9a1f56d6 100644 --- a/authentik/policies/utils.py +++ b/authentik/policies/utils.py @@ -1,8 +1,8 @@ """Policy Utils""" -from typing import Any, Dict +from typing import Any -def delete_none_keys(dict_: Dict[Any, Any]) -> Dict[Any, Any]: +def delete_none_keys(dict_: dict[Any, Any]) -> dict[Any, Any]: """Remove any keys from `dict_` that are None.""" new_dict = {} for key, value in dict_.items(): diff --git a/authentik/providers/oauth2/models.py b/authentik/providers/oauth2/models.py index e1fb623bc..885e20afa 100644 --- a/authentik/providers/oauth2/models.py +++ b/authentik/providers/oauth2/models.py @@ -5,7 +5,7 @@ import json import time from dataclasses import asdict, dataclass, field from hashlib import sha256 -from typing import Any, Dict, List, Optional, Type +from typing import Any, List, Optional, Type from urllib.parse import urlparse from uuid import uuid4 @@ -299,7 +299,7 @@ class OAuth2Provider(Provider): def __str__(self): return f"OAuth2 Provider {self.name}" - def encode(self, payload: Dict[str, Any]) -> str: + def encode(self, payload: dict[str, Any]) -> str: """Represent the ID Token as a JSON Web Token (JWT).""" keys = self.get_jwt_keys() # If the provider does not have an RSA Key assigned, it was switched to Symmetric @@ -394,9 +394,9 @@ class IDToken: nonce: Optional[str] = None at_hash: Optional[str] = None - claims: Dict[str, Any] = field(default_factory=dict) + claims: dict[str, Any] = field(default_factory=dict) - def to_dict(self) -> Dict[str, Any]: + def to_dict(self) -> dict[str, Any]: """Convert dataclass to dict, and update with keys from `claims`""" dic = asdict(self) dic.pop("claims") diff --git a/authentik/providers/oauth2/views/provider.py b/authentik/providers/oauth2/views/provider.py index cbe21316f..ab28e8db3 100644 --- a/authentik/providers/oauth2/views/provider.py +++ b/authentik/providers/oauth2/views/provider.py @@ -1,5 +1,5 @@ """authentik OAuth2 OpenID well-known views""" -from typing import Any, Dict +from typing import Any from django.http import HttpRequest, HttpResponse, JsonResponse from django.shortcuts import get_object_or_404, reverse @@ -29,7 +29,7 @@ PLAN_CONTEXT_SCOPES = "scopes" class ProviderInfoView(View): """OpenID-compliant Provider Info""" - def get_info(self, provider: OAuth2Provider) -> Dict[str, Any]: + def get_info(self, provider: OAuth2Provider) -> dict[str, Any]: """Get dictionary for OpenID Connect information""" scopes = list( ScopeMapping.objects.filter(provider=provider).values_list( diff --git a/authentik/providers/oauth2/views/session.py b/authentik/providers/oauth2/views/session.py index 332a9b87e..7d306188b 100644 --- a/authentik/providers/oauth2/views/session.py +++ b/authentik/providers/oauth2/views/session.py @@ -1,5 +1,5 @@ """authentik OAuth2 Session Views""" -from typing import Any, Dict +from typing import Any from django.shortcuts import get_object_or_404 from django.views.generic.base import TemplateView @@ -12,7 +12,7 @@ class EndSessionView(TemplateView): template_name = "providers/oauth2/end_session.html" - def get_context_data(self, **kwargs: Any) -> Dict[str, Any]: + def get_context_data(self, **kwargs: Any) -> dict[str, Any]: context = super().get_context_data(**kwargs) context["application"] = get_object_or_404( diff --git a/authentik/providers/oauth2/views/token.py b/authentik/providers/oauth2/views/token.py index 9b7433c73..c405a957b 100644 --- a/authentik/providers/oauth2/views/token.py +++ b/authentik/providers/oauth2/views/token.py @@ -2,7 +2,7 @@ from base64 import urlsafe_b64encode from dataclasses import InitVar, dataclass from hashlib import sha256 -from typing import Any, Dict, List, Optional +from typing import Any, List, Optional from django.http import HttpRequest, HttpResponse from django.views import View @@ -171,7 +171,7 @@ class TokenView(View): except UserAuthError as error: return TokenResponse(error.create_dict(), status=403) - def create_code_response_dic(self) -> Dict[str, Any]: + def create_code_response_dic(self) -> dict[str, Any]: """See https://tools.ietf.org/html/rfc6749#section-4.1""" refresh_token = self.params.authorization_code.provider.create_refresh_token( @@ -207,7 +207,7 @@ class TokenView(View): return response_dict - def create_refresh_response_dic(self) -> Dict[str, Any]: + def create_refresh_response_dic(self) -> dict[str, Any]: """See https://tools.ietf.org/html/rfc6749#section-6""" unauthorized_scopes = set(self.params.scope) - set( diff --git a/authentik/providers/oauth2/views/userinfo.py b/authentik/providers/oauth2/views/userinfo.py index 7ccdde78d..bda0e8f4e 100644 --- a/authentik/providers/oauth2/views/userinfo.py +++ b/authentik/providers/oauth2/views/userinfo.py @@ -1,5 +1,5 @@ """authentik OAuth2 OpenID Userinfo views""" -from typing import Any, Dict, List +from typing import Any, List from django.http import HttpRequest, HttpResponse from django.utils.translation import gettext_lazy as _ @@ -22,7 +22,7 @@ class UserInfoView(View): """Create a dictionary with all the requested claims about the End-User. See: http://openid.net/specs/openid-connect-core-1_0.html#UserInfoResponse""" - def get_scope_descriptions(self, scopes: List[str]) -> Dict[str, str]: + def get_scope_descriptions(self, scopes: List[str]) -> dict[str, str]: """Get a list of all Scopes's descriptions""" scope_descriptions = {} for scope in ScopeMapping.objects.filter(scope_name__in=scopes).order_by( @@ -47,7 +47,7 @@ class UserInfoView(View): scope_descriptions[scope] = github_scope_map[scope] return scope_descriptions - def get_claims(self, token: RefreshToken) -> Dict[str, Any]: + def get_claims(self, token: RefreshToken) -> dict[str, Any]: """Get a dictionary of claims from scopes that the token requires and are assigned to the provider.""" diff --git a/authentik/providers/proxy/controllers/docker.py b/authentik/providers/proxy/controllers/docker.py index e823696ac..b57adc06e 100644 --- a/authentik/providers/proxy/controllers/docker.py +++ b/authentik/providers/proxy/controllers/docker.py @@ -1,5 +1,4 @@ """Proxy Provider Docker Contoller""" -from typing import Dict from urllib.parse import urlparse from authentik.outposts.controllers.base import DeploymentPort @@ -18,7 +17,7 @@ class ProxyDockerController(DockerController): DeploymentPort(4443, "https", "tcp"), ] - def _get_labels(self) -> Dict[str, str]: + def _get_labels(self) -> dict[str, str]: hosts = [] for proxy_provider in ProxyProvider.objects.filter(outpost__in=[self.outpost]): proxy_provider: ProxyProvider diff --git a/authentik/providers/proxy/controllers/k8s/ingress.py b/authentik/providers/proxy/controllers/k8s/ingress.py index 94be2a04c..d7c69268a 100644 --- a/authentik/providers/proxy/controllers/k8s/ingress.py +++ b/authentik/providers/proxy/controllers/k8s/ingress.py @@ -1,5 +1,5 @@ """Kubernetes Ingress Reconciler""" -from typing import TYPE_CHECKING, Dict +from typing import TYPE_CHECKING from urllib.parse import urlparse from kubernetes.client import ( @@ -78,7 +78,7 @@ class IngressReconciler(KubernetesObjectReconciler[NetworkingV1beta1Ingress]): if have_hosts_tls != expected_hosts_tls: raise NeedsUpdate() - def get_ingress_annotations(self) -> Dict[str, str]: + def get_ingress_annotations(self) -> dict[str, str]: """Get ingress annotations""" annotations = { # Ensure that with multiple proxy replicas deployed, the same CSRF request diff --git a/authentik/root/asgi.py b/authentik/root/asgi.py index f8bd1fadf..81f75ab71 100644 --- a/authentik/root/asgi.py +++ b/authentik/root/asgi.py @@ -8,7 +8,7 @@ https://docs.djangoproject.com/en/3.0/howto/deployment/asgi/ """ import typing from time import time -from typing import Any, ByteString, Dict +from typing import Any, ByteString import django from asgiref.compatibility import guarantee_single_callable @@ -64,7 +64,7 @@ class ASGILogger: app: ASGIApp scope: Scope - headers: Dict[ByteString, Any] + headers: dict[ByteString, Any] status_code: int start: float diff --git a/authentik/sources/ldap/signals.py b/authentik/sources/ldap/signals.py index 5e7f8c3c7..08d94c07a 100644 --- a/authentik/sources/ldap/signals.py +++ b/authentik/sources/ldap/signals.py @@ -1,5 +1,5 @@ """authentik ldap source signals""" -from typing import Any, Dict +from typing import Any from django.core.exceptions import ValidationError from django.db.models.signals import post_save @@ -26,7 +26,7 @@ def sync_ldap_source_on_save(sender, instance: LDAPSource, **_): @receiver(password_validate) # pylint: disable=unused-argument -def ldap_password_validate(sender, password: str, plan_context: Dict[str, Any], **__): +def ldap_password_validate(sender, password: str, plan_context: dict[str, Any], **__): """if there's an LDAP Source with enabled password sync, check the password""" sources = LDAPSource.objects.filter(sync_users_password=True) if not sources.exists(): diff --git a/authentik/sources/oauth/clients/base.py b/authentik/sources/oauth/clients/base.py index 3d4e090c4..4b81eb50b 100644 --- a/authentik/sources/oauth/clients/base.py +++ b/authentik/sources/oauth/clients/base.py @@ -1,5 +1,5 @@ """OAuth Clients""" -from typing import Any, Dict, Optional +from typing import Any, Optional from urllib.parse import urlencode from django.http import HttpRequest @@ -33,11 +33,11 @@ class BaseOAuthClient: self.callback = callback self.session.headers.update({"User-Agent": f"authentik {__version__}"}) - def get_access_token(self, **request_kwargs) -> Optional[Dict[str, Any]]: + def get_access_token(self, **request_kwargs) -> Optional[dict[str, Any]]: "Fetch access token from callback request." raise NotImplementedError("Defined in a sub-class") # pragma: no cover - def get_profile_info(self, token: Dict[str, str]) -> Optional[Dict[str, Any]]: + def get_profile_info(self, token: dict[str, str]) -> Optional[dict[str, Any]]: "Fetch user profile information." try: response = self.do_request("get", self.source.profile_url, token=token) @@ -48,7 +48,7 @@ class BaseOAuthClient: else: return response.json() - def get_redirect_args(self) -> Dict[str, str]: + def get_redirect_args(self) -> dict[str, str]: "Get request parameters for redirect url." raise NotImplementedError("Defined in a sub-class") # pragma: no cover @@ -61,7 +61,7 @@ class BaseOAuthClient: LOGGER.info("redirect args", **args) return f"{self.source.authorization_url}?{params}" - def parse_raw_token(self, raw_token: str) -> Dict[str, Any]: + def parse_raw_token(self, raw_token: str) -> dict[str, Any]: "Parse token and secret from raw token response." raise NotImplementedError("Defined in a sub-class") # pragma: no cover diff --git a/authentik/sources/oauth/clients/oauth1.py b/authentik/sources/oauth/clients/oauth1.py index f558f18a2..191dc9b65 100644 --- a/authentik/sources/oauth/clients/oauth1.py +++ b/authentik/sources/oauth/clients/oauth1.py @@ -1,5 +1,5 @@ """OAuth 1 Clients""" -from typing import Any, Dict, Optional +from typing import Any, Optional from urllib.parse import parse_qsl from requests.exceptions import RequestException @@ -20,7 +20,7 @@ class OAuthClient(BaseOAuthClient): "Accept": "application/json", } - def get_access_token(self, **request_kwargs) -> Optional[Dict[str, Any]]: + def get_access_token(self, **request_kwargs) -> Optional[dict[str, Any]]: "Fetch access token from callback request." raw_token = self.request.session.get(self.session_key, None) verifier = self.request.GET.get("oauth_verifier", None) @@ -60,7 +60,7 @@ class OAuthClient(BaseOAuthClient): else: return response.text - def get_redirect_args(self) -> Dict[str, Any]: + def get_redirect_args(self) -> dict[str, Any]: "Get request parameters for redirect url." callback = self.request.build_absolute_uri(self.callback) raw_token = self.get_request_token() @@ -71,7 +71,7 @@ class OAuthClient(BaseOAuthClient): "oauth_callback": callback, } - def parse_raw_token(self, raw_token: str) -> Dict[str, Any]: + def parse_raw_token(self, raw_token: str) -> dict[str, Any]: "Parse token and secret from raw token response." return dict(parse_qsl(raw_token)) @@ -80,7 +80,7 @@ class OAuthClient(BaseOAuthClient): resource_owner_key = None resource_owner_secret = None if "token" in kwargs: - user_token: Dict[str, Any] = kwargs.pop("token") + user_token: dict[str, Any] = kwargs.pop("token") resource_owner_key = user_token["oauth_token"] resource_owner_secret = user_token["oauth_token_secret"] diff --git a/authentik/sources/oauth/clients/oauth2.py b/authentik/sources/oauth/clients/oauth2.py index f710db06f..968f919cc 100644 --- a/authentik/sources/oauth/clients/oauth2.py +++ b/authentik/sources/oauth/clients/oauth2.py @@ -1,6 +1,6 @@ """OAuth 2 Clients""" from json import loads -from typing import Any, Dict, Optional +from typing import Any, Optional from urllib.parse import parse_qsl from django.utils.crypto import constant_time_compare, get_random_string @@ -38,7 +38,7 @@ class OAuth2Client(BaseOAuthClient): "Generate state optional parameter." return get_random_string(32) - def get_access_token(self, **request_kwargs) -> Optional[Dict[str, Any]]: + def get_access_token(self, **request_kwargs) -> Optional[dict[str, Any]]: "Fetch access token from callback request." callback = self.request.build_absolute_uri(self.callback or self.request.path) if not self.check_application_state(): @@ -69,11 +69,11 @@ class OAuth2Client(BaseOAuthClient): else: return response.json() - def get_redirect_args(self) -> Dict[str, str]: + def get_redirect_args(self) -> dict[str, str]: "Get request parameters for redirect url." callback = self.request.build_absolute_uri(self.callback) client_id: str = self.source.consumer_key - args: Dict[str, str] = { + args: dict[str, str] = { "client_id": client_id, "redirect_uri": callback, "response_type": "code", @@ -84,7 +84,7 @@ class OAuth2Client(BaseOAuthClient): self.request.session[self.session_key] = state return args - def parse_raw_token(self, raw_token: str) -> Dict[str, Any]: + def parse_raw_token(self, raw_token: str) -> dict[str, Any]: "Parse token and secret from raw token response." # Load as json first then parse as query string try: diff --git a/authentik/sources/oauth/types/azure_ad.py b/authentik/sources/oauth/types/azure_ad.py index 4f370593e..7ca6468a2 100644 --- a/authentik/sources/oauth/types/azure_ad.py +++ b/authentik/sources/oauth/types/azure_ad.py @@ -1,5 +1,5 @@ """AzureAD OAuth2 Views""" -from typing import Any, Dict +from typing import Any from uuid import UUID from authentik.sources.oauth.models import OAuthSource, UserOAuthSourceConnection @@ -11,15 +11,15 @@ from authentik.sources.oauth.views.callback import OAuthCallback class AzureADOAuthCallback(OAuthCallback): """AzureAD OAuth2 Callback""" - def get_user_id(self, source: OAuthSource, info: Dict[str, Any]) -> str: + def get_user_id(self, source: OAuthSource, info: dict[str, Any]) -> str: return str(UUID(info.get("objectId")).int) def get_user_enroll_context( self, source: OAuthSource, access: UserOAuthSourceConnection, - info: Dict[str, Any], - ) -> Dict[str, Any]: + info: dict[str, Any], + ) -> dict[str, Any]: mail = info.get("mail", None) or info.get("otherMails", [None])[0] return { "username": info.get("displayName"), diff --git a/authentik/sources/oauth/types/discord.py b/authentik/sources/oauth/types/discord.py index 52fb412c6..94ab600e6 100644 --- a/authentik/sources/oauth/types/discord.py +++ b/authentik/sources/oauth/types/discord.py @@ -1,5 +1,5 @@ """Discord OAuth Views""" -from typing import Any, Dict +from typing import Any from authentik.sources.oauth.models import OAuthSource, UserOAuthSourceConnection from authentik.sources.oauth.types.manager import MANAGER, RequestKind @@ -25,8 +25,8 @@ class DiscordOAuth2Callback(OAuthCallback): self, source: OAuthSource, access: UserOAuthSourceConnection, - info: Dict[str, Any], - ) -> Dict[str, Any]: + info: dict[str, Any], + ) -> dict[str, Any]: return { "username": info.get("username"), "email": info.get("email", None), diff --git a/authentik/sources/oauth/types/facebook.py b/authentik/sources/oauth/types/facebook.py index 1f30c27e4..5b39c5265 100644 --- a/authentik/sources/oauth/types/facebook.py +++ b/authentik/sources/oauth/types/facebook.py @@ -1,5 +1,5 @@ """Facebook OAuth Views""" -from typing import Any, Dict, Optional +from typing import Any, Optional from facebook import GraphAPI @@ -23,7 +23,7 @@ class FacebookOAuthRedirect(OAuthRedirect): class FacebookOAuth2Client(OAuth2Client): """Facebook OAuth2 Client""" - def get_profile_info(self, token: Dict[str, str]) -> Optional[Dict[str, Any]]: + def get_profile_info(self, token: dict[str, str]) -> Optional[dict[str, Any]]: api = GraphAPI(access_token=token["access_token"]) return api.get_object("me", fields="id,name,email") @@ -38,8 +38,8 @@ class FacebookOAuth2Callback(OAuthCallback): self, source: OAuthSource, access: UserOAuthSourceConnection, - info: Dict[str, Any], - ) -> Dict[str, Any]: + info: dict[str, Any], + ) -> dict[str, Any]: return { "username": info.get("name"), "email": info.get("email"), diff --git a/authentik/sources/oauth/types/github.py b/authentik/sources/oauth/types/github.py index b0abbb49a..d016d6fcd 100644 --- a/authentik/sources/oauth/types/github.py +++ b/authentik/sources/oauth/types/github.py @@ -1,5 +1,5 @@ """GitHub OAuth Views""" -from typing import Any, Dict +from typing import Any from authentik.sources.oauth.models import OAuthSource, UserOAuthSourceConnection from authentik.sources.oauth.types.manager import MANAGER, RequestKind @@ -14,8 +14,8 @@ class GitHubOAuth2Callback(OAuthCallback): self, source: OAuthSource, access: UserOAuthSourceConnection, - info: Dict[str, Any], - ) -> Dict[str, Any]: + info: dict[str, Any], + ) -> dict[str, Any]: return { "username": info.get("login"), "email": info.get("email"), diff --git a/authentik/sources/oauth/types/google.py b/authentik/sources/oauth/types/google.py index 2a007b135..00143d324 100644 --- a/authentik/sources/oauth/types/google.py +++ b/authentik/sources/oauth/types/google.py @@ -1,5 +1,5 @@ """Google OAuth Views""" -from typing import Any, Dict +from typing import Any from authentik.sources.oauth.models import OAuthSource, UserOAuthSourceConnection from authentik.sources.oauth.types.manager import MANAGER, RequestKind @@ -25,8 +25,8 @@ class GoogleOAuth2Callback(OAuthCallback): self, source: OAuthSource, access: UserOAuthSourceConnection, - info: Dict[str, Any], - ) -> Dict[str, Any]: + info: dict[str, Any], + ) -> dict[str, Any]: return { "username": info.get("email"), "email": info.get("email"), diff --git a/authentik/sources/oauth/types/manager.py b/authentik/sources/oauth/types/manager.py index a74c67398..90b5317b9 100644 --- a/authentik/sources/oauth/types/manager.py +++ b/authentik/sources/oauth/types/manager.py @@ -1,6 +1,6 @@ """Source type manager""" from enum import Enum -from typing import Callable, Dict, List +from typing import Callable, List from django.utils.text import slugify from structlog.stdlib import get_logger @@ -22,7 +22,7 @@ class RequestKind(Enum): class SourceTypeManager: """Manager to hold all Source types.""" - __source_types: Dict[RequestKind, Dict[str, Callable]] = {} + __source_types: dict[RequestKind, dict[str, Callable]] = {} __names: List[str] = [] def source(self, kind: RequestKind, name: str): diff --git a/authentik/sources/oauth/types/oidc.py b/authentik/sources/oauth/types/oidc.py index 2555c1270..00d4723e1 100644 --- a/authentik/sources/oauth/types/oidc.py +++ b/authentik/sources/oauth/types/oidc.py @@ -1,5 +1,5 @@ """OpenID Connect OAuth Views""" -from typing import Any, Dict +from typing import Any from authentik.sources.oauth.models import OAuthSource, UserOAuthSourceConnection from authentik.sources.oauth.types.manager import MANAGER, RequestKind @@ -21,15 +21,15 @@ class OpenIDConnectOAuthRedirect(OAuthRedirect): class OpenIDConnectOAuth2Callback(OAuthCallback): """OpenIDConnect OAuth2 Callback""" - def get_user_id(self, source: OAuthSource, info: Dict[str, str]) -> str: + def get_user_id(self, source: OAuthSource, info: dict[str, str]) -> str: return info.get("sub", "") def get_user_enroll_context( self, source: OAuthSource, access: UserOAuthSourceConnection, - info: Dict[str, Any], - ) -> Dict[str, Any]: + info: dict[str, Any], + ) -> dict[str, Any]: return { "username": info.get("nickname"), "email": info.get("email"), diff --git a/authentik/sources/oauth/types/reddit.py b/authentik/sources/oauth/types/reddit.py index 4c60ece56..d032642ff 100644 --- a/authentik/sources/oauth/types/reddit.py +++ b/authentik/sources/oauth/types/reddit.py @@ -1,5 +1,5 @@ """Reddit OAuth Views""" -from typing import Any, Dict +from typing import Any from requests.auth import HTTPBasicAuth @@ -40,8 +40,8 @@ class RedditOAuth2Callback(OAuthCallback): self, source: OAuthSource, access: UserOAuthSourceConnection, - info: Dict[str, Any], - ) -> Dict[str, Any]: + info: dict[str, Any], + ) -> dict[str, Any]: return { "username": info.get("name"), "email": None, diff --git a/authentik/sources/oauth/types/twitter.py b/authentik/sources/oauth/types/twitter.py index 3b7acf6cf..ba90237bd 100644 --- a/authentik/sources/oauth/types/twitter.py +++ b/authentik/sources/oauth/types/twitter.py @@ -1,5 +1,5 @@ """Twitter OAuth Views""" -from typing import Any, Dict +from typing import Any from authentik.sources.oauth.models import OAuthSource, UserOAuthSourceConnection from authentik.sources.oauth.types.manager import MANAGER, RequestKind @@ -14,8 +14,8 @@ class TwitterOAuthCallback(OAuthCallback): self, source: OAuthSource, access: UserOAuthSourceConnection, - info: Dict[str, Any], - ) -> Dict[str, Any]: + info: dict[str, Any], + ) -> dict[str, Any]: return { "username": info.get("screen_name"), "email": info.get("email", None), diff --git a/authentik/sources/oauth/views/callback.py b/authentik/sources/oauth/views/callback.py index cac0b05ec..17d3ad8bd 100644 --- a/authentik/sources/oauth/views/callback.py +++ b/authentik/sources/oauth/views/callback.py @@ -1,5 +1,5 @@ """OAuth Callback Views""" -from typing import Any, Dict, Optional +from typing import Any, Optional from django.conf import settings from django.contrib import messages @@ -115,14 +115,14 @@ class OAuthCallback(OAuthClientMixin, View): self, source: OAuthSource, access: UserOAuthSourceConnection, - info: Dict[str, Any], - ) -> Dict[str, Any]: + info: dict[str, Any], + ) -> dict[str, Any]: """Create a dict of User data""" raise NotImplementedError() # pylint: disable=unused-argument def get_user_id( - self, source: UserOAuthSourceConnection, info: Dict[str, Any] + self, source: UserOAuthSourceConnection, info: dict[str, Any] ) -> Optional[str]: """Return unique identifier from the profile info.""" if "id" in info: @@ -167,7 +167,7 @@ class OAuthCallback(OAuthClientMixin, View): source: OAuthSource, user: User, access: UserOAuthSourceConnection, - info: Dict[str, Any], + info: dict[str, Any], ) -> HttpResponse: "Login user and redirect." messages.success( @@ -184,7 +184,7 @@ class OAuthCallback(OAuthClientMixin, View): self, source: OAuthSource, access: UserOAuthSourceConnection, - info: Dict[str, Any], + info: dict[str, Any], ) -> HttpResponse: """Handler when the user was already authenticated and linked an external source to their account.""" @@ -211,7 +211,7 @@ class OAuthCallback(OAuthClientMixin, View): self, source: OAuthSource, access: UserOAuthSourceConnection, - info: Dict[str, Any], + info: dict[str, Any], ) -> HttpResponse: """User was not authenticated and previous request was not authenticated.""" messages.success( diff --git a/authentik/sources/oauth/views/redirect.py b/authentik/sources/oauth/views/redirect.py index aa0919946..39f1faa8b 100644 --- a/authentik/sources/oauth/views/redirect.py +++ b/authentik/sources/oauth/views/redirect.py @@ -1,5 +1,5 @@ """OAuth Redirect Views""" -from typing import Any, Dict +from typing import Any from django.http import Http404 from django.urls import reverse @@ -19,7 +19,7 @@ class OAuthRedirect(OAuthClientMixin, RedirectView): params = None # pylint: disable=unused-argument - def get_additional_parameters(self, source: OAuthSource) -> Dict[str, Any]: + def get_additional_parameters(self, source: OAuthSource) -> dict[str, Any]: "Return additional redirect parameters for this source." return self.params or {} diff --git a/authentik/sources/saml/processors/request.py b/authentik/sources/saml/processors/request.py index 7c9cbd7f3..e16ca2777 100644 --- a/authentik/sources/saml/processors/request.py +++ b/authentik/sources/saml/processors/request.py @@ -1,6 +1,5 @@ """SAML AuthnRequest Processor""" from base64 import b64encode -from typing import Dict from urllib.parse import quote_plus import xmlsec @@ -125,7 +124,7 @@ class RequestProcessor: return etree.tostring(auth_n_request).decode() - def build_auth_n_detached(self) -> Dict[str, str]: + def build_auth_n_detached(self) -> dict[str, str]: """Get Dict AuthN Request for Redirect bindings, with detached Signature. See https://docs.oasis-open.org/security/saml/v2.0/saml-bindings-2.0-os.pdf""" auth_n_request = self.get_auth_n() diff --git a/authentik/sources/saml/processors/response.py b/authentik/sources/saml/processors/response.py index f99e188c6..00eff2f3d 100644 --- a/authentik/sources/saml/processors/response.py +++ b/authentik/sources/saml/processors/response.py @@ -1,6 +1,6 @@ """authentik saml source processor""" from base64 import b64decode -from typing import TYPE_CHECKING, Any, Dict +from typing import TYPE_CHECKING, Any import xmlsec from defusedxml.lxml import fromstring @@ -154,7 +154,7 @@ class ResponseProcessor: raise ValueError("NameID Element not found!") return name_id - def _get_name_id_filter(self) -> Dict[str, str]: + def _get_name_id_filter(self) -> dict[str, str]: """Returns the subject's NameID as a Filter for the `User`""" name_id_el = self._get_name_id() name_id = name_id_el.text diff --git a/authentik/stages/authenticator_static/stage.py b/authentik/stages/authenticator_static/stage.py index f02778297..839ad8afb 100644 --- a/authentik/stages/authenticator_static/stage.py +++ b/authentik/stages/authenticator_static/stage.py @@ -1,5 +1,5 @@ """Static OTP Setup stage""" -from typing import Any, Dict +from typing import Any from django.http import HttpRequest, HttpResponse from django.views.generic import FormView @@ -21,7 +21,7 @@ class AuthenticatorStaticStageView(FormView, StageView): form_class = SetupForm - def get_form_kwargs(self, **kwargs) -> Dict[str, Any]: + def get_form_kwargs(self, **kwargs) -> dict[str, Any]: kwargs = super().get_form_kwargs(**kwargs) tokens = self.request.session[SESSION_STATIC_TOKENS] kwargs["tokens"] = tokens diff --git a/authentik/stages/authenticator_totp/stage.py b/authentik/stages/authenticator_totp/stage.py index 631b24cf6..cfada422b 100644 --- a/authentik/stages/authenticator_totp/stage.py +++ b/authentik/stages/authenticator_totp/stage.py @@ -1,5 +1,5 @@ """TOTP Setup stage""" -from typing import Any, Dict +from typing import Any from django.http import HttpRequest, HttpResponse from django.utils.encoding import force_str @@ -24,7 +24,7 @@ class AuthenticatorTOTPStageView(FormView, StageView): form_class = SetupForm - def get_form_kwargs(self, **kwargs) -> Dict[str, Any]: + def get_form_kwargs(self, **kwargs) -> dict[str, Any]: kwargs = super().get_form_kwargs(**kwargs) device: TOTPDevice = self.request.session[SESSION_TOTP_DEVICE] kwargs["device"] = device diff --git a/authentik/stages/authenticator_validate/stage.py b/authentik/stages/authenticator_validate/stage.py index 04c3018b2..0fd969ba3 100644 --- a/authentik/stages/authenticator_validate/stage.py +++ b/authentik/stages/authenticator_validate/stage.py @@ -1,5 +1,5 @@ """OTP Validation""" -from typing import Any, Dict +from typing import Any from django.http import HttpRequest, HttpResponse from django.views.generic import FormView @@ -20,7 +20,7 @@ class AuthenticatorValidateStageView(FormView, StageView): form_class = ValidationForm - def get_form_kwargs(self, **kwargs) -> Dict[str, Any]: + def get_form_kwargs(self, **kwargs) -> dict[str, Any]: kwargs = super().get_form_kwargs(**kwargs) kwargs["user"] = self.executor.plan.context.get(PLAN_CONTEXT_PENDING_USER) return kwargs diff --git a/authentik/stages/consent/stage.py b/authentik/stages/consent/stage.py index c4f2feea4..009f1e836 100644 --- a/authentik/stages/consent/stage.py +++ b/authentik/stages/consent/stage.py @@ -1,5 +1,5 @@ """authentik consent stage""" -from typing import Any, Dict, List +from typing import Any, List from django.http import HttpRequest, HttpResponse from django.utils.timezone import now @@ -19,7 +19,7 @@ class ConsentStageView(FormView, StageView): form_class = ConsentForm - def get_context_data(self, **kwargs: Dict[str, Any]) -> Dict[str, Any]: + def get_context_data(self, **kwargs: dict[str, Any]) -> dict[str, Any]: kwargs = super().get_context_data(**kwargs) kwargs["current_stage"] = self.executor.current_stage kwargs["context"] = self.executor.plan.context diff --git a/authentik/stages/dummy/stage.py b/authentik/stages/dummy/stage.py index 9cedfa479..fefb2c0dc 100644 --- a/authentik/stages/dummy/stage.py +++ b/authentik/stages/dummy/stage.py @@ -1,5 +1,5 @@ """authentik multi-stage authentication engine""" -from typing import Any, Dict +from typing import Any from django.http import HttpRequest @@ -13,7 +13,7 @@ class DummyStageView(StageView): """Just redirect to next stage""" return self.executor.stage_ok() - def get_context_data(self, **kwargs: Dict[str, Any]) -> Dict[str, Any]: + def get_context_data(self, **kwargs: dict[str, Any]) -> dict[str, Any]: kwargs = super().get_context_data(**kwargs) kwargs["title"] = self.executor.current_stage.name return kwargs diff --git a/authentik/stages/password/stage.py b/authentik/stages/password/stage.py index 39c2df528..946b74eb1 100644 --- a/authentik/stages/password/stage.py +++ b/authentik/stages/password/stage.py @@ -1,5 +1,5 @@ """authentik password stage""" -from typing import Any, Dict, List, Optional +from typing import Any, List, Optional from django.contrib.auth import _clean_credentials from django.contrib.auth.backends import BaseBackend @@ -24,7 +24,7 @@ SESSION_INVALID_TRIES = "user_invalid_tries" def authenticate( - request: HttpRequest, backends: List[str], **credentials: Dict[str, Any] + request: HttpRequest, backends: List[str], **credentials: dict[str, Any] ) -> Optional[User]: """If the given credentials are valid, return a User object. diff --git a/authentik/stages/prompt/signals.py b/authentik/stages/prompt/signals.py index 24e4f332b..a98f249ba 100644 --- a/authentik/stages/prompt/signals.py +++ b/authentik/stages/prompt/signals.py @@ -1,5 +1,5 @@ """authentik prompt stage signals""" from django.core.signals import Signal -# Arguments: password: str, plan_context: Dict[str, Any] +# Arguments: password: str, plan_context: dict[str, Any] password_validate = Signal() diff --git a/authentik/stages/user_write/signals.py b/authentik/stages/user_write/signals.py index 7a4c3811a..34e7cfdba 100644 --- a/authentik/stages/user_write/signals.py +++ b/authentik/stages/user_write/signals.py @@ -1,5 +1,5 @@ """authentik user_write signals""" from django.core.signals import Signal -# Arguments: request: HttpRequest, user: User, data: Dict[str, Any], created: bool +# Arguments: request: HttpRequest, user: User, data: dict[str, Any], created: bool user_write = Signal() diff --git a/tests/e2e/test_flows_enroll.py b/tests/e2e/test_flows_enroll.py index 4fb42aaff..553c7a05c 100644 --- a/tests/e2e/test_flows_enroll.py +++ b/tests/e2e/test_flows_enroll.py @@ -1,6 +1,6 @@ """Test Enroll flow""" from sys import platform -from typing import Any, Dict, Optional +from typing import Any, Optional from unittest.case import skipUnless from django.test import override_settings @@ -22,7 +22,7 @@ from tests.e2e.utils import USER, SeleniumTestCase, retry class TestFlowsEnroll(SeleniumTestCase): """Test Enroll flow""" - def get_container_specs(self) -> Optional[Dict[str, Any]]: + def get_container_specs(self) -> Optional[dict[str, Any]]: return { "image": "mailhog/mailhog:v1.0.1", "detach": True, diff --git a/tests/e2e/test_provider_oauth2_github.py b/tests/e2e/test_provider_oauth2_github.py index 3a1ebd36d..aae6c11c2 100644 --- a/tests/e2e/test_provider_oauth2_github.py +++ b/tests/e2e/test_provider_oauth2_github.py @@ -1,7 +1,7 @@ """test OAuth Provider flow""" from sys import platform from time import sleep -from typing import Any, Dict, Optional +from typing import Any, Optional from unittest.case import skipUnless from docker.types import Healthcheck @@ -30,7 +30,7 @@ class TestProviderOAuth2Github(SeleniumTestCase): self.client_secret = generate_client_secret() super().setUp() - def get_container_specs(self) -> Optional[Dict[str, Any]]: + def get_container_specs(self) -> Optional[dict[str, Any]]: """Setup client grafana container which we test OAuth against""" return { "image": "grafana/grafana:7.1.0", diff --git a/tests/e2e/test_provider_oauth2_grafana.py b/tests/e2e/test_provider_oauth2_grafana.py index e7a4de640..cd2dc95a5 100644 --- a/tests/e2e/test_provider_oauth2_grafana.py +++ b/tests/e2e/test_provider_oauth2_grafana.py @@ -1,7 +1,7 @@ """test OAuth2 OpenID Provider flow""" from sys import platform from time import sleep -from typing import Any, Dict, Optional +from typing import Any, Optional from unittest.case import skipUnless from docker.types import Healthcheck @@ -40,7 +40,7 @@ class TestProviderOAuth2OAuth(SeleniumTestCase): self.client_secret = generate_client_secret() super().setUp() - def get_container_specs(self) -> Optional[Dict[str, Any]]: + def get_container_specs(self) -> Optional[dict[str, Any]]: return { "image": "grafana/grafana:7.1.0", "detach": True, diff --git a/tests/e2e/test_provider_proxy.py b/tests/e2e/test_provider_proxy.py index 74be9c218..cbbb1ee12 100644 --- a/tests/e2e/test_provider_proxy.py +++ b/tests/e2e/test_provider_proxy.py @@ -2,7 +2,7 @@ from dataclasses import asdict from sys import platform from time import sleep -from typing import Any, Dict, Optional +from typing import Any, Optional from unittest.case import skipUnless from channels.testing import ChannelsLiveServerTestCase @@ -35,7 +35,7 @@ class TestProviderProxy(SeleniumTestCase): super().tearDown() self.proxy_container.kill() - def get_container_specs(self) -> Optional[Dict[str, Any]]: + def get_container_specs(self) -> Optional[dict[str, Any]]: return { "image": "traefik/whoami:latest", "detach": True, diff --git a/tests/e2e/test_source_oauth.py b/tests/e2e/test_source_oauth.py index 9c0f6f4ab..9f72366a2 100644 --- a/tests/e2e/test_source_oauth.py +++ b/tests/e2e/test_source_oauth.py @@ -2,7 +2,7 @@ from os.path import abspath from sys import platform from time import sleep -from typing import Any, Dict, Optional +from typing import Any, Optional from unittest.case import skipUnless from django.test import override_settings @@ -72,7 +72,7 @@ class TestSourceOAuth2(SeleniumTestCase): with open(CONFIG_PATH, "w+") as _file: safe_dump(config, _file) - def get_container_specs(self) -> Optional[Dict[str, Any]]: + def get_container_specs(self) -> Optional[dict[str, Any]]: return { "image": "quay.io/dexidp/dex:v2.24.0", "detach": True, @@ -249,7 +249,7 @@ class TestSourceOAuth1(SeleniumTestCase): self.source_slug = "oauth1-test" super().setUp() - def get_container_specs(self) -> Optional[Dict[str, Any]]: + def get_container_specs(self) -> Optional[dict[str, Any]]: return { "image": "beryju/oauth1-test-server", "detach": True, diff --git a/tests/e2e/test_source_saml.py b/tests/e2e/test_source_saml.py index e43b07e3e..037eac95e 100644 --- a/tests/e2e/test_source_saml.py +++ b/tests/e2e/test_source_saml.py @@ -1,7 +1,7 @@ """test SAML Source""" from sys import platform from time import sleep -from typing import Any, Dict, Optional +from typing import Any, Optional from unittest.case import skipUnless from docker.types import Healthcheck @@ -73,7 +73,7 @@ Sm75WXsflOxuTn08LbgGc4s= class TestSourceSAML(SeleniumTestCase): """test SAML Source flow""" - def get_container_specs(self) -> Optional[Dict[str, Any]]: + def get_container_specs(self) -> Optional[dict[str, Any]]: return { "image": "kristophjunge/test-saml-idp:1.15", "detach": True, diff --git a/tests/e2e/utils.py b/tests/e2e/utils.py index c2acbe28a..e63ff2a9d 100644 --- a/tests/e2e/utils.py +++ b/tests/e2e/utils.py @@ -6,7 +6,7 @@ from importlib.util import module_from_spec, spec_from_file_location from inspect import getmembers, isfunction from os import environ, makedirs from time import sleep, time -from typing import Any, Callable, Dict, Optional +from typing import Any, Callable, Optional from django.apps import apps from django.contrib.staticfiles.testing import StaticLiveServerTestCase @@ -56,7 +56,7 @@ class SeleniumTestCase(StaticLiveServerTestCase): if specs := self.get_container_specs(): self.container = self._start_container(specs) - def _start_container(self, specs: Dict[str, Any]) -> Container: + def _start_container(self, specs: dict[str, Any]) -> Container: client: DockerClient = from_env() client.images.pull(specs["image"]) container = client.containers.run(**specs) @@ -70,7 +70,7 @@ class SeleniumTestCase(StaticLiveServerTestCase): self.logger.info("Container failed healthcheck") sleep(1) - def get_container_specs(self) -> Optional[Dict[str, Any]]: + def get_container_specs(self) -> Optional[dict[str, Any]]: """Optionally get container specs which will launched on setup, wait for the container to be healthy, and deleted again on tearDown""" return None