Coverage for paperless/adapter.py: 38%
79 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 09:07 +0000
1import logging
2from urllib.parse import quote
4from allauth.account.adapter import DefaultAccountAdapter
5from allauth.core import context
6from allauth.headless.tokens.strategies.sessions import SessionTokenStrategy
7from allauth.socialaccount.adapter import DefaultSocialAccountAdapter
8from django.conf import settings
9from django.contrib.auth.models import Group
10from django.contrib.auth.models import User
11from django.forms import ValidationError
12from django.http import HttpRequest
13from django.urls import reverse
14from rest_framework.authtoken.models import Token
16from documents.models import Document
17from paperless.signals import handle_social_account_updated
19logger = logging.getLogger("paperless.auth")
22class CustomAccountAdapter(DefaultAccountAdapter):
23 def is_open_for_signup(self, request):
24 """
25 Check whether the site is open for signups, which can be
26 disabled via the ACCOUNT_ALLOW_SIGNUPS setting.
27 """
28 if (
29 User.objects.exclude(username__in=["consumer", "AnonymousUser"]).count()
30 == 0
31 and Document.global_objects.count() == 0
32 ):
33 # I.e. a fresh install, allow signups
34 return True
35 allow_signups = super().is_open_for_signup(request)
36 # Override with setting, otherwise default to super.
37 return getattr(settings, "ACCOUNT_ALLOW_SIGNUPS", allow_signups)
39 def pre_authenticate(self, request, **credentials):
40 """
41 Called prior to calling the authenticate method on the
42 authentication backend. If login is disabled using DISABLE_REGULAR_LOGIN,
43 raise ValidationError to prevent the login.
44 """
45 if settings.DISABLE_REGULAR_LOGIN:
46 raise ValidationError("Regular login is disabled")
48 return super().pre_authenticate(request, **credentials)
50 def is_safe_url(self, url):
51 """
52 Check if the URL is a safe URL.
53 See https://github.com/paperless-ngx/paperless-ngx/issues/5780
54 """
55 from django.utils.http import url_has_allowed_host_and_scheme
57 # get_host already validates the given host, so no need to check it again
58 allowed_hosts = {context.request.get_host()} | set(settings.ALLOWED_HOSTS)
60 if "*" in allowed_hosts: 60 ↛ 66line 60 didn't jump to line 66 because the condition on line 60 was always true
61 # dont allow wildcard to allow urls from any host
62 allowed_hosts.remove("*")
63 allowed_hosts.add(context.request.get_host())
64 return url_has_allowed_host_and_scheme(url, allowed_hosts=allowed_hosts)
66 return url_has_allowed_host_and_scheme(url, allowed_hosts=allowed_hosts)
68 def get_reset_password_from_key_url(self, key):
69 """
70 Return the URL to reset a password e.g. in reset email.
71 """
72 if settings.PAPERLESS_URL is None:
73 return super().get_reset_password_from_key_url(key)
74 else:
75 path = reverse(
76 "account_reset_password_from_key",
77 kwargs={"uidb36": "UID", "key": "KEY"},
78 )
79 path = path.replace("UID-KEY", quote(key))
80 return settings.PAPERLESS_URL + path
82 def save_user(self, request, user, form, commit=True): # noqa: FBT002
83 """
84 Save the user instance. Default groups are assigned to the user, if
85 specified in the settings.
86 """
88 if (
89 User.objects.exclude(username__in=["consumer", "AnonymousUser"]).count()
90 == 0
91 and Document.global_objects.count() == 0
92 ):
93 # I.e. a fresh install, make the user a superuser
94 logger.debug(f"Creating initial superuser `{user}`")
95 user.is_superuser = True
96 user.is_staff = True
98 user: User = super().save_user(request, user, form, commit)
99 group_names: list[str] = settings.ACCOUNT_DEFAULT_GROUPS
100 if len(group_names) > 0:
101 groups = Group.objects.filter(name__in=group_names)
102 logger.debug(f"Adding default groups to user `{user}`: {group_names}")
103 user.groups.add(*groups)
104 user.save()
105 return user
108class CustomSocialAccountAdapter(DefaultSocialAccountAdapter):
109 def is_open_for_signup(self, request, sociallogin):
110 """
111 Check whether the site is open for signups via social account, which can be
112 disabled via the SOCIALACCOUNT_ALLOW_SIGNUPS setting.
113 """
114 allow_signups = super().is_open_for_signup(request, sociallogin)
115 # Override with setting, otherwise default to super.
116 return getattr(settings, "SOCIALACCOUNT_ALLOW_SIGNUPS", allow_signups)
118 def get_connect_redirect_url(self, request, socialaccount):
119 """
120 Returns the default URL to redirect to after successfully
121 connecting a social account.
122 """
123 url = reverse("base")
124 return url
126 def save_user(self, request, sociallogin, form=None):
127 """
128 Save the user instance. Default groups are assigned to the user, if
129 specified in the settings.
130 """
131 # save_user also calls account_adapter save_user which would set ACCOUNT_DEFAULT_GROUPS
132 user: User = super().save_user(request, sociallogin, form)
133 group_names: list[str] = settings.SOCIAL_ACCOUNT_DEFAULT_GROUPS
134 if len(group_names) > 0:
135 groups = Group.objects.filter(name__in=group_names)
136 logger.debug(
137 f"Adding default social groups to user `{user}`: {group_names}",
138 )
139 user.groups.add(*groups)
140 user.save()
141 handle_social_account_updated(None, request, sociallogin)
142 return user
144 def on_authentication_error(
145 self,
146 request,
147 provider,
148 error=None,
149 exception=None,
150 extra_context=None,
151 ):
152 """
153 Just log errors and pass them along.
154 """
155 logger.warning(
156 f"Social authentication error for provider `{provider!s}`: {error!s} ({exception!s})",
157 )
158 return super().on_authentication_error(
159 request,
160 provider,
161 error,
162 exception,
163 extra_context,
164 )
167class DrfTokenStrategy(SessionTokenStrategy):
168 def create_access_token(self, request: HttpRequest) -> str | None:
169 if not request.user.is_authenticated:
170 return None
171 token, _ = Token.objects.get_or_create(user=request.user)
172 return token.key