Coverage for src/backend/InvenTree/InvenTree/auth_overrides.py: 27%

1"""Overrides for allauth and adjacent packages to enforce InvenTree specific auth settings and restirctions.""" 

2 

3from typing import Literal 

4 

5from django import forms 

6from django.conf import settings 

7from django.contrib.auth.models import Group 

8from django.core.exceptions import PermissionDenied 

9from django.http import HttpResponseRedirect 

10from django.urls import reverse 

11from django.utils.translation import gettext_lazy as _ 

12 

13import structlog 

14from allauth.account.adapter import DefaultAccountAdapter 

15from allauth.account.forms import LoginForm, SignupForm, set_form_field_order 

16from allauth.headless.adapter import DefaultHeadlessAdapter 

17from allauth.socialaccount.adapter import DefaultSocialAccountAdapter 

18 

19from common.settings import get_global_setting 

20from InvenTree.exceptions import log_error 

21 

22from .helpers import str2bool 

23from .helpers_email import is_email_configured 

24 

25logger = structlog.get_logger('inventree') 

26 

27 

28# override allauth 

29class CustomLoginForm(LoginForm): 

30 """Custom login form to override default allauth behaviour.""" 

31 

32 def login(self, request, redirect_url=None): 

33 """Perform login action. 

34 

35 First check that: 

36 - A valid user has been supplied 

37 """ 

38 if not self.user: 

39 # No user supplied - redirect to the login page 

40 return HttpResponseRedirect(reverse('account_login')) 

41 

42 # Now perform default login action 

43 return super().login(request, redirect_url) 

44 

45 

46class CustomSignupForm(SignupForm): 

47 """Override to use dynamic settings.""" 

48 

49 def __init__(self, *args, **kwargs): 

50 """Check settings to influence which fields are needed.""" 

51 kwargs['email_required'] = get_global_setting('LOGIN_MAIL_REQUIRED') 

52 

53 super().__init__(*args, **kwargs) 

54 

55 # check for two mail fields 

56 if get_global_setting('LOGIN_SIGNUP_MAIL_TWICE'): 

57 self.fields['email2'] = forms.EmailField( 

58 label=_('Email (again)'), 

59 widget=forms.TextInput( 

60 attrs={ 

61 'type': 'email', 

62 'placeholder': _('Email address confirmation'), 

63 } 

64 ), 

65 ) 

66 

67 # check for two password fields 

68 if not get_global_setting('LOGIN_SIGNUP_PWD_TWICE'): 

69 self.fields.pop('password2', None) 

70 

71 # reorder fields 

72 set_form_field_order( 

73 self, ['username', 'email', 'email2', 'password1', 'password2'] 

74 ) 

75 

76 def clean(self): 

77 """Make sure the supplied emails match if enabled in settings.""" 

78 cleaned_data = super().clean() 

79 

80 # check for two mail fields 

81 if get_global_setting('LOGIN_SIGNUP_MAIL_TWICE'): 

82 email = cleaned_data.get('email') 

83 email2 = cleaned_data.get('email2') 

84 if (email and email2) and email != email2: 

85 self.add_error('email2', _('You must type the same email each time.')) 

86 

87 return cleaned_data 

88 

89 

90RegistrationKeys = Literal['LOGIN_ENABLE_REG', 'LOGIN_ENABLE_SSO_REG'] 

91 

92 

93def registration_enabled(setting_name: RegistrationKeys = 'LOGIN_ENABLE_REG'): 

94 """Determine whether user registration is enabled.""" 

95 if str2bool(get_global_setting(setting_name)): 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true

96 if is_email_configured(): 

97 return True 

98 else: 

99 logger.warning( 

100 'INVE-W11: Registration cannot be enabled, because EMAIL_HOST is not configured.' 

101 ) 

102 return False 

103 

104 

105class RegistrationMixin: 

106 """Mixin to check if registration should be enabled.""" 

107 

108 REGISTRATION_SETTING: RegistrationKeys = 'LOGIN_ENABLE_REG' 

109 

110 def is_open_for_signup(self, request, *args, **kwargs): 

111 """Check if signup is enabled in settings. 

112 

113 Configure the class variable `REGISTRATION_SETTING` to set which setting should be used, default: `LOGIN_ENABLE_REG`. 

114 """ 

115 if registration_enabled(self.REGISTRATION_SETTING): 

116 return True 

117 # Only warn when this is an actual signup submission, not when called as 

118 # a feature-availability check during login or other auth flows. 

119 if request and request.method == 'POST' and 'signup' in request.path: 

120 logger.warning( 

121 f'INVE-W12: Signup attempt blocked, because registration is disabled via setting {self.REGISTRATION_SETTING}.' 

122 ) 

123 return False 

124 

125 def clean_email(self, email): 

126 """Check if the mail is valid to the pattern in LOGIN_SIGNUP_MAIL_RESTRICTION (if enabled in settings).""" 

127 mail_restriction = get_global_setting('LOGIN_SIGNUP_MAIL_RESTRICTION', None) 

128 if not mail_restriction: 

129 return super().clean_email(email) 

130 

131 split_email = email.split('@') 

132 if len(split_email) != 2: 

133 logger.error('The user %s has an invalid email address', email) 

134 raise forms.ValidationError( 

135 _('The provided primary email address is not valid.') 

136 ) 

137 

138 mailoptions = mail_restriction.split(',') 

139 for option in mailoptions: 

140 if not option.startswith('@'): 

141 raise forms.ValidationError( 

142 _('The provided primary email address is not valid.') 

143 ) 

144 elif split_email[1] == option[1:]: 

145 return super().clean_email(email) 

146 

147 logger.info('The provided email domain for %s is not approved', email) 

148 raise forms.ValidationError(_('The provided email domain is not approved.')) 

149 

150 def save_user(self, request, user, form, commit=True): 

151 """Check if a default group is set in settings.""" 

152 # Create the user 

153 user = super().save_user(request, user, form) 

154 

155 # Check if a default group is set in settings 

156 start_group = get_global_setting('SIGNUP_GROUP') 

157 if ( 

158 start_group and user.groups.count() == 0 

159 ): # check that no group has been added through SSO group sync 

160 try: 

161 group = Group.objects.get(id=start_group) 

162 user.groups.add(group) 

163 except Group.DoesNotExist: 

164 logger.exception( 

165 'The setting `SIGNUP_GROUP` contains an non existent group', 

166 start_group, 

167 ) 

168 user.save() 

169 return user 

170 

171 

172class CustomAccountAdapter(RegistrationMixin, DefaultAccountAdapter): 

173 """Override of adapter to use dynamic settings.""" 

174 

175 def send_mail(self, template_prefix, email, context): 

176 """Only send mail if backend configured.""" 

177 if settings.EMAIL_HOST: 

178 try: 

179 result = super().send_mail(template_prefix, email, context) 

180 except Exception: 

181 # An exception occurred while attempting to send email 

182 # Log it (for admin users) and return silently 

183 log_error('send_mail', scope='auth') 

184 result = False 

185 

186 return result 

187 

188 return False 

189 

190 def send_password_reset_mail(self, user, email, context): 

191 """Send the password reset mail.""" 

192 if not get_global_setting('LOGIN_ENABLE_PWD_FORGOT'): 

193 raise PermissionDenied('Password reset is disabled') 

194 return super().send_password_reset_mail(user, email, context) 

195 

196 

197class CustomSocialAccountAdapter(RegistrationMixin, DefaultSocialAccountAdapter): 

198 """Override of adapter to use dynamic settings.""" 

199 

200 REGISTRATION_SETTING = 'LOGIN_ENABLE_SSO_REG' 

201 

202 def is_auto_signup_allowed(self, request, sociallogin): 

203 """Check if auto signup is enabled in settings.""" 

204 if get_global_setting('LOGIN_SIGNUP_SSO_AUTO', True): 

205 return super().is_auto_signup_allowed(request, sociallogin) 

206 return False 

207 

208 def authentication_error( 

209 self, request, provider_id, error=None, exception=None, extra_context=None 

210 ): 

211 """Callback method for authentication errors.""" 

212 if not error: 

213 error = request.GET.get('error', None) 

214 

215 if not exception: 

216 exception = request.GET.get('error_description', None) 

217 

218 path = request.path or 'sso' 

219 

220 # Log the error to the database 

221 log_error(path, error_name=error, error_data=exception, scope='auth') 

222 logger.error("SSO error for provider '%s' - check admin error log", provider_id) 

223 

224 def get_connect_redirect_url(self, request, socialaccount): 

225 """Redirect to the frontend after connecting an account.""" 

226 return request.build_absolute_uri(f'/{settings.FRONTEND_URL_BASE}/') 

227 

228 

229class CustomHeadlessAdapter(DefaultHeadlessAdapter): 

230 """Override of adapter to use dynamic settings.""" 

231 

232 def get_frontend_url(self, urlname, **kwargs): 

233 """Get the frontend URL for the given URL name respecting the request.""" 

234 HEADLESS_FRONTEND_URLS = { 

235 'account_confirm_email': 'verify-email/{key}', 

236 'account_reset_password': 'reset-password', 

237 'account_reset_password_from_key': 'set-password?key={key}', 

238 'account_signup': 'register', 

239 'socialaccount_login_error': 'social-login-error', 

240 } 

241 if urlname not in HEADLESS_FRONTEND_URLS: 

242 raise ValueError( 

243 f'URL name "{urlname}" not found in HEADLESS_FRONTEND_URLS' 

244 ) # pragma: no cover 

245 

246 return self.request.build_absolute_uri( 

247 f'/{settings.FRONTEND_URL_BASE}/{HEADLESS_FRONTEND_URLS[urlname].format(**kwargs)}' 

248 )