Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/modules/user.py: 35%
800 statements
« prev ^ index » next coverage.py v7.17.2, created at 2026-11-12 02:43 +0000
1import abc
2import datetime
3import enum
4import fnmatch
5import hashlib
6import hmac
7import json
8import logging
9import re
10import secrets
11import time
12import urllib.parse
13import warnings
14from http.cookies import SimpleCookie
16import user_agents
18import pyotp
19import base64
20import dataclasses
21import typing as t
22from google.auth.transport import requests
23from google.oauth2 import id_token
25from viur.core import (
26 conf, current, db, email, errors, i18n,
27 securitykey, session, skeleton, tasks, utils, Module
28)
29from viur.core.decorators import *
30from viur.core.bones import *
31from viur.core.bones.password import PBKDF2_DEFAULT_ITERATIONS, encode_password
32from viur.core.prototypes.list import List
33from viur.core.ratelimit import RateLimit
34from viur.core.securityheaders import extendCsp
35from viur.core.session import Session
38class Status(enum.IntEnum):
39 """Status enum for a user
41 This is an IntEnum, so it is comparable with plain ints as well as with
42 other IntEnum classes of the same values, e.g. when a project defines its
43 own Status enum to add custom status values to a subclassed UserSkel:
45 class Status(enum.IntEnum):
46 UNSET = 0
47 WAITING_FOR_EMAIL_VERIFICATION = 1
48 WAITING_FOR_ADMIN_VERIFICATION = 2
49 DISABLED = 5
50 ACTIVE = 10
51 PENDING_REVIEW = 15 # custom, project-specific status
53 class UserSkel(user.UserSkel):
54 status = SelectBone(
55 ...,
56 values=Status,
57 defaultValue=Status.ACTIVE,
58 )
59 """
61 UNSET = 0 # Status is unset
62 WAITING_FOR_EMAIL_VERIFICATION = 1 # Waiting for email verification
63 WAITING_FOR_ADMIN_VERIFICATION = 2 # Waiting for verification through admin
64 DISABLED = 5 # Account disabled
65 ACTIVE = 10 # Active
68class UserSkel(skeleton.Skeleton):
69 kindName = "user" # this assignment is required, as this Skeleton is defined in viur-core (see #604)
71 name = EmailBone(
72 descr="E-Mail",
73 required=True,
74 readOnly=True,
75 caseSensitive=False,
76 searchable=True,
77 unique=UniqueValue(UniqueLockMethod.SameValue, True, "Username already taken"),
78 tags=("personal", "identifier", "contact"),
79 )
81 firstname = StringBone(
82 descr="Firstname",
83 searchable=True,
84 tags="personal",
85 )
87 lastname = StringBone(
88 descr="Lastname",
89 searchable=True,
90 tags="personal",
91 )
93 roles = SelectBone(
94 descr=i18n.translate("viur.core.modules.user.bone.roles", defaultText="Roles"),
95 values=conf.user.roles,
96 required=True,
97 multiple=True,
98 # fixme: This is generally broken in VIUR! See #776 for details.
99 # vfunc=lambda values:
100 # i18n.translate(
101 # "user.bone.roles.invalid",
102 # defaultText="Invalid role setting: 'custom' can only be set alone.")
103 # if "custom" in values and len(values) > 1 else None,
104 defaultValue=list(conf.user.roles.keys())[:1],
105 )
107 access = SelectBone(
108 descr=i18n.translate("viur.core.modules.user.bone.access", defaultText="Access rights"),
109 type_suffix="access",
110 values=lambda: {
111 right: i18n.translate(f"viur.core.modules.user.accessright.{right}", defaultText=right)
112 for right in sorted(conf.user.access_rights)
113 },
114 multiple=True,
115 params={
116 "readonlyIf": "'custom' not in roles" # if "custom" is not in roles, "access" is managed by the role system
117 },
118 )
120 status = SelectBone(
121 descr=i18n.translate("viur.core.modules.user.bone.status", "Account status"),
122 values=Status,
123 translation_key_prefix="viur.core.user.status.",
124 defaultValue=Status.ACTIVE,
125 required=True,
126 )
128 lastlogin = DateBone(
129 descr="Last Login",
130 readOnly=True,
131 tags=("personal", "technical"),
132 )
134 admin_config = JsonBone( # This bone stores settings from the admin
135 descr="Config for the User",
136 visible=False,
137 )
139 def __new__(cls, *args, **kwargs):
140 """
141 Constructor for the UserSkel-class, with the capability
142 to dynamically add bones required for the configured
143 authentication methods.
144 """
145 for provider in conf.main_app.vi.user.authenticationProviders:
146 assert issubclass(provider, UserPrimaryAuthentication)
147 provider.patch_user_skel(cls)
149 for provider in conf.main_app.vi.user.secondFactorProviders:
150 assert issubclass(provider, UserSecondFactorAuthentication)
151 provider.patch_user_skel(cls)
153 cls.__boneMap__ = skeleton.MetaBaseSkel.generate_bonemap(cls)
154 return super().__new__(cls, *args, **kwargs)
156 @classmethod
157 def write(cls, skel, *args, **kwargs):
158 # Roles
159 if skel["roles"] and "custom" not in skel["roles"]:
160 # Collect access rights through rules
161 access = set()
163 for role in skel["roles"]:
164 # Get default access for this role
165 access |= conf.main_app.vi.user.get_role_defaults(role)
167 # Go through all modules and evaluate available role-settings
168 for name in dir(conf.main_app.vi):
169 if name.startswith("_"):
170 continue
172 module = getattr(conf.main_app.vi, name)
173 if not isinstance(module, Module):
174 continue
176 roles = getattr(module, "roles", None) or {}
177 rights = roles.get(role, roles.get("*", ()))
179 # Convert role into tuple if it's not
180 if not isinstance(rights, (tuple, list)):
181 rights = (rights, )
183 if "*" in rights:
184 for right in module.accessRights:
185 access.add(f"{name}-{right}")
186 else:
187 for right in rights:
188 if right in module.accessRights:
189 access.add(f"{name}-{right}")
191 # special case: "edit" and "delete" actions require "view" as well!
192 if right in ("edit", "delete") and "view" in module.accessRights:
193 access.add(f"{name}-view")
195 skel["access"] = list(access)
197 return super().write(skel, *args, **kwargs)
200class UserAuthentication(Module, abc.ABC):
201 @property
202 @abc.abstractstaticmethod
203 def METHOD_NAME() -> str:
204 """
205 Define a unique method name for this authentication.
206 """
207 ...
209 @property
210 @abc.abstractstaticmethod
211 def NAME() -> str:
212 """
213 Define a descriptive name for this authentication.
214 """
215 ...
217 @property
218 @staticmethod
219 def VISIBLE(cls) -> bool:
220 """
221 Defines if the authentication method is visible to the user.
222 """
223 return True
225 def __init__(self, moduleName, modulePath, userModule):
226 super().__init__(moduleName, modulePath)
227 self._user_module = userModule
228 self.start_url = f"{self.modulePath}/login"
230 def can_handle(self, skel: skeleton.SkeletonInstance) -> bool:
231 return True
233 @classmethod
234 def patch_user_skel(cls, skel_cls: skeleton.Skeleton):
235 """
236 Allows for an UserAuthentication to patch the UserSkel
237 class with additional bones which are required for
238 the implemented authentication method.
239 """
240 ...
243class UserPrimaryAuthentication(UserAuthentication, abc.ABC):
244 """Abstract class for all primary authentication methods."""
245 registrationEnabled = False
247 @abc.abstractmethod
248 def login(self, *args, **kwargs):
249 ...
251 def next_or_finish(self, skel: skeleton.SkeletonInstance):
252 """
253 Hook that is called whenever a part of the authentication was successful.
254 It allows to perform further steps in custom authentications,
255 e.g. change a password after first login.
256 """
257 return self._user_module.continueAuthenticationFlow(self, skel["key"])
260class UserPassword(UserPrimaryAuthentication):
261 METHOD_NAME = "X-VIUR-AUTH-User-Password"
262 NAME = "Username & Password"
264 registrationEmailVerificationRequired = True
265 registrationAdminVerificationRequired = True
267 verifySuccessTemplate = "user_verify_success"
268 verifyEmailAddressMail = "user_verify_address"
269 verifyFailedTemplate = "user_verify_failed"
270 passwordRecoveryMail = "user_password_recovery"
271 passwordRecoverySuccessTemplate = "user_passwordrecover_success"
272 passwordRecoveryTemplate = "user_passwordrecover"
274 # The default rate-limit for password recovery (10 tries each 15 minutes)
275 passwordRecoveryRateLimit = RateLimit("user.passwordrecovery", 10, 15, "ip")
277 # Limit (invalid) login-retries to once per 5 seconds
278 loginRateLimit = RateLimit("user.login", 12, 1, "ip")
280 @classmethod
281 def patch_user_skel(cls, skel_cls):
282 """
283 Modifies the UserSkel to be equipped by a PasswordBone.
284 """
285 skel_cls.password = PasswordBone(
286 readOnly=True,
287 visible=False,
288 params={
289 "category": "Authentication",
290 },
291 )
293 class LoginSkel(skeleton.RelSkel):
294 name = EmailBone(
295 descr="E-Mail",
296 required=True,
297 caseSensitive=False,
298 )
299 password = PasswordBone(
300 required=True,
301 test_threshold=0,
302 tests=(),
303 raw=True,
304 )
306 class LostPasswordStep1Skel(skeleton.RelSkel):
307 name = EmailBone(
308 descr="E-Mail",
309 required=True,
310 )
312 class LostPasswordStep2Skel(skeleton.RelSkel):
313 recovery_key = StringBone(
314 descr="Recovery Key",
315 required=True,
316 params={
317 "tooltip": i18n.translate(
318 key="viur.core.modules.user.userpassword.lostpasswordstep2.recoverykey",
319 defaultText="Please enter the validation key you've received via e-mail.",
320 hint="Shown when the user needs more than 15 minutes to paste the key",
321 ),
322 }
323 )
325 class LostPasswordStep3Skel(skeleton.RelSkel):
326 # send the recovery key again, in case the password is rejected by some reason.
327 recovery_key = StringBone(
328 descr="Recovery Key",
329 visible=False,
330 )
332 password = PasswordBone(
333 descr="New Password",
334 required=True,
335 params={
336 "tooltip": i18n.translate(
337 key="viur.core.modules.user.userpassword.lostpasswordstep3.password",
338 defaultText="Please enter a new password for your account.",
339 ),
340 }
341 )
343 @exposed
344 @force_ssl
345 @skey(allow_empty=True)
346 def login(self, **kwargs):
347 # Obtain a fresh login skel
348 skel = self.LoginSkel()
350 # Read required bones from client
351 if not (kwargs and skel.fromClient(kwargs)):
352 return self._user_module.render.render("login", skel)
354 self.loginRateLimit.assertQuotaIsAvailable()
356 # query for the username. The query might find another user, but the name is being checked for equality below
357 name = skel["name"].lower().strip()
358 user_skel = self._user_module.baseSkel()
359 user_skel = user_skel.all().filter("name.idx >=", name).getSkel() or user_skel
361 # extract password hash from raw database entity (skeleton access blocks it)
362 password_data = (user_skel.dbEntity and user_skel.dbEntity.get("password")) or {}
363 iterations = password_data.get("iterations", 1001) # remember iterations; old password hashes used 1001
364 password_hash = encode_password(skel["password"], password_data.get("salt", "-invalid-"), iterations)["pwhash"]
366 # now check if the username matches
367 is_okay = secrets.compare_digest((user_skel["name"] or "").lower().strip().encode(), name.encode())
369 # next, check if the password hash matches
370 is_okay &= secrets.compare_digest(password_data.get("pwhash", b"-invalid-"), password_hash)
372 if not is_okay:
373 # Set error to all required fields
374 for name, bone in skel.items():
375 if bone.required:
376 skel.errors.append(
377 ReadFromClientError(
378 ReadFromClientErrorSeverity.Invalid,
379 i18n.translate(
380 key="viur.core.modules.user.userpassword.login.failed",
381 defaultText="Invalid username or password provided",
382 ),
383 name,
384 )
385 )
387 self.loginRateLimit.decrementQuota() # Only failed login attempts will count to the quota
388 return self._user_module.render.render("login", skel)
390 # check if iterations are below current security standards, and update if necessary.
391 if iterations < PBKDF2_DEFAULT_ITERATIONS:
392 logging.info(f"Update password hash for user {name}.")
393 # re-hash the password with more iterations
394 # FIXME: This must be done within a transaction!
395 user_skel["password"] = kwargs["password"] # will be hashed on serialize
396 user_skel.write(update_relations=False)
398 return self.next_or_finish(user_skel)
400 @exposed
401 def pwrecover(self, recovery_key: str | None = None, skey: str | None = None, **kwargs):
402 """
403 This implements a password recovery process which lets users set a new password for their account,
404 after validating a recovery key sent by email.
406 The process is as following:
408 - The user enters the registered email adress (not validated here)
409 - A random code is generated and stored as a security-ke, then sendUserPasswordRecoveryCode is called.
410 - sendUserPasswordRecoveryCode will run in the background, check if we have a user with that name
411 and send a link with the code. It runs as a deferred task so no information if a user account exists
412 is being leaked.
413 - If the user received an email, the link can be clicked to set a new password for the account.
415 To prevent automated attacks, the first step is guarded by limited calls to this function to 10 actions
416 per 15 minutes. (One complete recovery process consists of two calls).
417 """
418 self.passwordRecoveryRateLimit.assertQuotaIsAvailable()
419 current_request = current.request.get()
421 if recovery_key is None:
422 # This is the first step, where we ask for the username of the account we'll going to reset the password on
423 skel = self.LostPasswordStep1Skel()
425 if (
426 not kwargs
427 or not current_request.isPostRequest
428 or not skel.fromClient(kwargs)
429 ):
430 return self._user_module.render.render(
431 "pwrecover", skel,
432 tpl=self.passwordRecoveryTemplate,
433 )
435 # validate security key
436 if not securitykey.validate(skey):
437 raise errors.PreconditionFailed()
439 self.passwordRecoveryRateLimit.decrementQuota()
441 recovery_key = securitykey.create(
442 duration=datetime.timedelta(minutes=15),
443 key_length=conf.security.password_recovery_key_length,
444 user_name=skel["name"].lower(),
445 session_bound=False,
446 )
448 # Send the code in background
449 self.sendUserPasswordRecoveryCode(
450 skel["name"], recovery_key, current_request.request.headers["User-Agent"]
451 )
453 # step 2 is only an action-skel, and can be ignored by a direct link in the
454 # e-mail previously sent. It depends on the implementation of the specific project.
455 return self._user_module.render.render(
456 "pwrecover", self.LostPasswordStep2Skel(),
457 tpl=self.passwordRecoveryTemplate,
458 )
460 # in step 3
461 skel = self.LostPasswordStep3Skel()
463 # reset the recovery key again, in case the fromClient() fails.
464 skel["recovery_key"] = str(recovery_key).strip()
466 # check for any input; Render input-form again when incomplete.
467 if (
468 not kwargs
469 or not current_request.isPostRequest
470 or not skel.fromClient(kwargs, ignore=("recovery_key",))
471 ):
472 return self._user_module.render.render(
473 "pwrecover", skel,
474 tpl=self.passwordRecoveryTemplate,
475 )
477 # validate security key
478 if not securitykey.validate(skey):
479 raise errors.PreconditionFailed()
481 if not (recovery_request := securitykey.validate(recovery_key, session_bound=False)):
482 raise errors.PreconditionFailed(
483 i18n.translate(
484 key="viur.core.modules.user.passwordrecovery.keyexpired",
485 defaultText="The recovery key is expired or invalid. Please start the recovery process again.",
486 hint="Shown when the user needs more than 15 minutes to paste the key, or entered an invalid key."
487 )
488 )
490 self.passwordRecoveryRateLimit.decrementQuota()
492 # If we made it here, the key was correct, so we'd hopefully have a valid user for this
493 user_skel = self._user_module.viewSkel().all().filter("name.idx =", recovery_request["user_name"]).getSkel()
495 if not user_skel:
496 raise errors.NotFound(
497 i18n.translate(
498 key="viur.core.modules.user.passwordrecovery.usernotfound",
499 defaultText="There is no account with this name",
500 hint="We cant find an account with that name (Should never happen)"
501 )
502 )
504 # If the account is locked or not yet validated, abort the process.
505 if not self._user_module.is_active(user_skel):
506 raise errors.NotFound(
507 i18n.translate(
508 key="viur.core.modules.user.passwordrecovery.accountlocked",
509 defaultText="This account is currently locked. You cannot change its password.",
510 hint="Attempted password recovery on a locked account"
511 )
512 )
514 # Update the password, save the user, reset his session and show the success-template
515 user_skel["password"] = skel["password"]
516 user_skel.write(update_relations=False)
518 return self._user_module.render.render(
519 "pwrecover_success",
520 next_url=self.start_url,
521 tpl=self.passwordRecoverySuccessTemplate
522 )
524 @tasks.CallDeferred
525 def sendUserPasswordRecoveryCode(self, user_name: str, recovery_key: str, user_agent: str) -> None:
526 """
527 Sends the given recovery code to the user given in userName. This function runs deferred
528 so there's no timing sidechannel that leaks if this user exists. Per default, we'll send the
529 code by email (assuming we have working email delivery), but this can be overridden to send it
530 by SMS or other means. We'll also update the changedate for this user, so no more than one code
531 can be send to any given user in four hours.
532 """
533 if user_skel := self._user_module.viewSkel().all().filter("name.idx =", user_name).getSkel():
534 user_agent = user_agents.parse(user_agent)
535 email.send_email(
536 tpl=self.passwordRecoveryMail,
537 skel=user_skel,
538 dests=[user_name],
539 recovery_key=recovery_key,
540 user_agent={
541 "device": user_agent.get_device(),
542 "os": user_agent.get_os(),
543 "browser": user_agent.get_browser()
544 }
545 )
547 @exposed
548 @skey(forward_payload="data", session_bound=False)
549 def verify(self, data):
550 def transact(key):
551 skel = self._user_module.editSkel()
552 if not key or not skel.read(key):
553 return None
555 skel["status"] = Status.WAITING_FOR_ADMIN_VERIFICATION \
556 if self.registrationAdminVerificationRequired else Status.ACTIVE
558 skel.write(update_relations=False)
559 return skel
561 if not isinstance(data, dict) or not (skel := db.run_in_transaction(transact, data.get("user_key"))):
562 return self._user_module.render.view(None, tpl=self.verifyFailedTemplate)
564 return self._user_module.render.view(skel, tpl=self.verifySuccessTemplate)
566 def canAdd(self) -> bool:
567 return self.registrationEnabled
569 def addSkel(self) -> skeleton.SkeletonInstance["UserSkel"]:
570 """
571 Prepare the add-Skel for rendering.
572 Currently only calls self._user_module.addSkel() and sets skel["status"] depending on
573 self.registrationEmailVerificationRequired and self.registrationAdminVerificationRequired
574 :return: viur.core.skeleton.Skeleton
575 """
576 skel = self._user_module.addSkel()
578 if self.registrationEmailVerificationRequired:
579 defaultStatusValue = Status.WAITING_FOR_EMAIL_VERIFICATION
580 elif self.registrationAdminVerificationRequired:
581 defaultStatusValue = Status.WAITING_FOR_ADMIN_VERIFICATION
582 else: # No further verification required
583 defaultStatusValue = Status.ACTIVE
585 skel.status.readOnly = True
586 skel["status"] = defaultStatusValue
588 if "password" in skel:
589 skel.password.required = True # The user will have to set a password
591 return skel
593 @force_ssl
594 @exposed
595 @skey(allow_empty=True)
596 def add(self, *, bounce: bool = False, **kwargs):
597 """
598 Allows guests to register a new account if self.registrationEnabled is set to true
600 .. seealso:: :func:`addSkel`, :func:`onAdded`, :func:`canAdd`, :func:`onAdd`
602 :returns: The rendered, added object of the entry, eventually with error hints.
604 :raises: :exc:`viur.core.errors.Unauthorized`, if the current user does not have the required permissions.
605 :raises: :exc:`viur.core.errors.PreconditionFailed`, if the *skey* could not be verified.
606 """
607 if not self.canAdd():
608 raise errors.Unauthorized()
610 skel = self.addSkel()
612 if (
613 not kwargs # no data supplied
614 or not current.request.get().isPostRequest # bail out if not using POST-method
615 or not skel.fromClient(kwargs) # failure on reading into the bones
616 or bounce # review before adding
617 ):
618 # render the skeleton in the version it could as far as it could be read.
619 return self._user_module.render.add(skel)
621 self._user_module.onAdd(skel)
622 skel.write()
624 if self.registrationEmailVerificationRequired and skel["status"] == Status.WAITING_FOR_EMAIL_VERIFICATION:
625 # The user will have to verify his email-address. Create a skey and send it to his address
626 skey = securitykey.create(duration=datetime.timedelta(days=7), session_bound=False,
627 user_key=db.normalize_key(skel["key"]),
628 name=skel["name"])
629 skel.skey = BaseBone(descr="Skey")
630 skel["skey"] = skey
631 email.send_email(dests=[skel["name"]], tpl=self.verifyEmailAddressMail, skel=skel)
633 self._user_module.onAdded(skel) # Call onAdded on our parent user module
634 return self._user_module.render.addSuccess(skel)
637class GoogleAccount(UserPrimaryAuthentication):
638 METHOD_NAME = "X-VIUR-AUTH-Google-Account"
639 NAME = "Google Account"
641 @classmethod
642 def patch_user_skel(cls, skel_cls):
643 """
644 Modifies the UserSkel to be equipped by a bones required by Google Auth
645 """
646 skel_cls.uid = StringBone(
647 descr="Google UserID",
648 required=False,
649 readOnly=True,
650 unique=UniqueValue(UniqueLockMethod.SameValue, False, "UID already in use"),
651 params={
652 "category": "Authentication",
653 },
654 )
656 skel_cls.sync = BooleanBone(
657 descr="Sync user data with OAuth-based services",
658 defaultValue=True,
659 params={
660 "category": "Authentication",
661 "tooltip":
662 "If set, user data like firstname and lastname is automatically kept"
663 "synchronous with the information stored at the OAuth service provider"
664 "(e.g. Google Login)."
665 },
666 )
668 @exposed
669 @force_ssl
670 @skey(allow_empty=True)
671 def login(self, token: str | None = None, *args, **kwargs):
672 if not conf.user.google_client_id:
673 raise errors.PreconditionFailed("Please configure conf.user.google_client_id!")
675 if not token:
676 request = current.request.get()
677 request.response.headers["Content-Type"] = "text/html"
678 if request.response.headers.get("cross-origin-opener-policy") == "same-origin":
679 # We have to allow popups here
680 request.response.headers["cross-origin-opener-policy"] = "same-origin-allow-popups"
682 file_path = conf.instance.core_base_path.joinpath("viur/core/template/vi_user_google_login.html")
683 with open(file_path) as file:
684 tpl_string = file.read()
686 # FIXME: Use Jinja2 for rendering?
687 tpl_string = tpl_string.replace("{{ clientID }}", conf.user.google_client_id)
688 extendCsp({
689 "script-src": ["sha256-JpzaUIxV/gVOQhKoDLerccwqDDIVsdn1JclA6kRNkLw="],
690 "style-src": ["sha256-FQpGSicYMVC5jxKGS5sIEzrRjSJmkxKPaetUc7eamqc="]
691 })
692 return tpl_string
694 user_info = id_token.verify_oauth2_token(token, requests.Request(), conf.user.google_client_id)
695 if user_info["iss"] not in {"accounts.google.com", "https://accounts.google.com"}:
696 raise ValueError("Invalid issuer")
698 # Token looks valid :)
699 uid = user_info["sub"]
700 email = user_info["email"]
702 base_skel = self._user_module.baseSkel()
703 update = False
704 if not (user_skel := base_skel.all().filter("uid =", uid).getSkel()):
705 # We'll try again - checking if there's already an user with that email
706 if not (user_skel := base_skel.all().filter("name.idx =", email.lower()).getSkel()):
707 # Still no luck - it's a completely new user
708 if not self.registrationEnabled:
709 if (domain := user_info.get("hd")) and domain in conf.user.google_gsuite_domains:
710 logging.debug(f"Google user is from allowed {domain} - adding account")
711 else:
712 logging.debug(f"Google user is from {domain} - denying registration")
713 raise errors.Forbidden("Registration for new users is disabled")
715 user_skel = base_skel
716 user_skel["uid"] = uid
717 user_skel["name"] = email
718 update = True
720 # Take user information from Google, if wanted!
721 if user_skel["sync"]:
722 for target, source in {
723 "name": email,
724 "firstname": user_info.get("given_name"),
725 "lastname": user_info.get("family_name"),
726 }.items():
728 if user_skel[target] != source:
729 user_skel[target] = source
730 update = True
732 if update:
733 assert user_skel.write()
735 return self.next_or_finish(user_skel)
738class UserSecondFactorAuthentication(UserAuthentication, abc.ABC):
739 """Abstract class for all second factors."""
740 MAX_RETRY = 3
741 second_factor_login_template = "user_login_secondfactor"
742 """Template to enter the TOPT on login"""
744 @property
745 @abc.abstractmethod
746 def NAME(self) -> str:
747 """Name for this factor for templates."""
748 ...
750 @property
751 @abc.abstractmethod
752 def ACTION_NAME(self) -> str:
753 """The action name for this factor, used as path-segment."""
754 ...
756 def __init__(self, moduleName, modulePath, _user_module):
757 super().__init__(moduleName, modulePath, _user_module)
758 self.action_url = f"{self.modulePath}/{self.ACTION_NAME}"
759 self.add_url = f"{self.modulePath}/add"
760 self.start_url = f"{self.modulePath}/start"
763class TimeBasedOTP(UserSecondFactorAuthentication):
764 METHOD_NAME = "X-VIUR-2FACTOR-TimeBasedOTP"
765 WINDOW_SIZE = 5
766 ACTION_NAME = "otp"
767 NAME = "Time-based OTP"
768 second_factor_login_template = "user_login_secondfactor"
770 @dataclasses.dataclass
771 class OtpConfig:
772 """
773 This dataclass is used to provide an interface for a OTP token
774 algorithm description that is passed within the TimeBasedOTP
775 class for configuration.
776 """
777 secret: str
778 timedrift: float = 0.0
779 algorithm: t.Literal["sha1", "sha256"] = "sha1"
780 interval: int = 60
782 class OtpSkel(skeleton.RelSkel):
783 """
784 This is the Skeleton used to ask for the OTP token.
785 """
786 otptoken = NumericBone(
787 descr="Token",
788 required=True,
789 max=999999,
790 min=0,
791 )
793 @classmethod
794 def patch_user_skel(cls, skel_cls):
795 """
796 Modifies the UserSkel to be equipped by a bones required by Timebased OTP
797 """
798 # One-Time Password Verification
799 skel_cls.otp_serial = StringBone(
800 descr="OTP serial",
801 searchable=True,
802 params={
803 "category": "Second Factor Authentication",
804 },
805 )
807 skel_cls.otp_secret = CredentialBone(
808 descr="OTP secret",
809 params={
810 "category": "Second Factor Authentication",
811 },
812 )
814 skel_cls.otp_timedrift = NumericBone(
815 descr="OTP time drift",
816 readOnly=True,
817 defaultValue=0,
818 precision=1,
819 params={
820 "category": "Second Factor Authentication",
821 },
822 )
824 def get_config(self, skel: skeleton.SkeletonInstance) -> OtpConfig | None:
825 """
826 Returns an instance of self.OtpConfig with a provided token configuration,
827 or None when there is no appropriate configuration of this second factor handler available.
828 """
830 if otp_secret := skel.dbEntity.get("otp_secret"):
831 return self.OtpConfig(secret=otp_secret, timedrift=skel.dbEntity.get("otp_timedrift") or 0)
833 return None
835 def can_handle(self, skel: skeleton.SkeletonInstance) -> bool:
836 """
837 Specified whether the second factor authentication can be handled by the given user or not.
838 """
839 return bool(self.get_config(skel))
841 @exposed
842 def start(self):
843 """
844 Configures OTP login for the current session.
846 A special otp_user_conf has to be specified as a dict, which is stored into the session.
847 """
848 session = current.session.get()
850 if not (user_key := session.get("possible_user_key")):
851 raise errors.PreconditionFailed(
852 "Second factor can only be triggered after successful primary authentication."
853 )
855 user_skel = self._user_module.baseSkel()
856 if not user_skel.read(user_key):
857 raise errors.NotFound("The previously authenticated user is gone.")
859 if not (otp_user_conf := self.get_config(user_skel)):
860 raise errors.PreconditionFailed("This second factor is not available for the user")
862 otp_user_conf = {
863 "key": str(user_key),
864 } | dataclasses.asdict(otp_user_conf)
866 session = current.session.get()
867 session["_otp_user"] = otp_user_conf
868 session.markChanged()
870 return self._user_module.render.edit(
871 self.OtpSkel(),
872 params={
873 "name": i18n.translate(
874 f"viur.core.modules.user.{self.ACTION_NAME}",
875 default_variables={"name": self.NAME},
876 ),
877 "action_name": self.ACTION_NAME,
878 "action_url": f"{self.modulePath}/{self.ACTION_NAME}",
879 },
880 tpl=self.second_factor_login_template
881 )
883 @exposed
884 @force_ssl
885 @skey(allow_empty=True)
886 def otp(self, *args, **kwargs):
887 """
888 Performs the second factor validation and interaction with the client.
889 """
890 session = current.session.get()
891 if not (otp_user_conf := session.get("_otp_user")): 891 ↛ 892line 891 didn't jump to line 892 because the condition on line 891 was never true
892 raise errors.PreconditionFailed("No OTP process started in this session")
894 # Check if maximum second factor verification attempts
895 if (attempts := otp_user_conf.get("attempts") or 0) > self.MAX_RETRY: 895 ↛ 896line 895 didn't jump to line 896 because the condition on line 895 was never true
896 raise errors.Forbidden("Maximum amount of authentication retries exceeded")
898 # Read the OTP token via the skeleton, to obtain a valid value
899 skel = self.OtpSkel()
900 if skel.fromClient(kwargs): 900 ↛ 911line 900 didn't jump to line 911 because the condition on line 900 was always true
901 # Verify the otptoken. If valid, this returns the current timedrift index for this hardware OTP.
902 res = self.verify(
903 otp=skel["otptoken"],
904 secret=otp_user_conf["secret"],
905 algorithm=otp_user_conf.get("algorithm") or "sha1",
906 interval=otp_user_conf.get("interval") or 60,
907 timedrift=otp_user_conf.get("timedrift") or 0.0,
908 valid_window=self.WINDOW_SIZE
909 )
910 else:
911 res = None
913 # Check if Token is invalid. Caution: 'if not verifyIndex' gets false positive for verifyIndex === 0!
914 if res is None: 914 ↛ 932line 914 didn't jump to line 932 because the condition on line 914 was always true
915 otp_user_conf["attempts"] = attempts + 1
916 session.markChanged()
917 skel.errors = [ReadFromClientError(ReadFromClientErrorSeverity.Invalid, "Wrong OTP Token", ["otptoken"])]
918 return self._user_module.render.edit(
919 skel,
920 params={
921 "name": i18n.translate(
922 f"viur.core.modules.user.auth.{self.ACTION_NAME}",
923 default_variables={"name": self.NAME},
924 ),
925 "action_name": self.ACTION_NAME,
926 "action_url": f"{self.modulePath}/{self.ACTION_NAME}",
927 },
928 tpl=self.second_factor_login_template
929 )
931 # Remove otp user config from session
932 user_key = db.key_helper(otp_user_conf["key"], self._user_module._resolveSkelCls().kindName)
933 del session["_otp_user"]
934 session.markChanged()
936 # Check if the OTP device has a time drift
938 timedriftchange = float(res) - otp_user_conf["timedrift"]
939 if abs(timedriftchange) > 2:
940 # The time-drift change accumulates to more than 2 minutes (for interval==60):
941 # update clock-drift value accordingly
942 self.updateTimeDrift(user_key, timedriftchange)
944 # Continue with authentication
945 return self._user_module.secondFactorSucceeded(self, user_key)
947 @staticmethod
948 def verify(
949 otp: str | int,
950 secret: str,
951 algorithm: str = "sha1",
952 interval: int = 60,
953 timedrift: float = 0.0,
954 for_time: datetime.datetime | None = None,
955 valid_window: int = 0,
956 ) -> int | None:
957 """
958 Verifies the OTP passed in against the current time OTP.
960 This is a fork of pyotp.verify. Rather than true/false, if valid_window > 0, it returns the index for which
961 the OTP value obtained by pyotp.at(for_time=time.time(), counter_offset=index) equals the current value shown
962 on the hardware token generator. This can be used to store the time drift of a given token generator.
964 :param otp: the OTP token to check against
965 :param secret: The OTP secret
966 :param algorithm: digest function to use in the HMAC (expected to be sha1 or sha256)
967 :param interval: the time interval in seconds for OTP. This defaults to 60 (old OTP c200 Generators).
968 :param timedrift: The known timedrift (old index) of the hardware OTP generator
969 :param for_time: Time to check OTP at (defaults to now)
970 :param valid_window: extends the validity to this many counter ticks before and after the current one
972 :returns: The index where verification succeeded, None otherwise
973 """
974 # get the hashing digest
975 digest = {
976 "sha1": hashlib.sha1,
977 "sha256": hashlib.sha256,
978 }.get(algorithm)
980 if not digest:
981 raise errors.NotImplemented(f"{algorithm=} is not implemented")
983 if for_time is None:
984 for_time = datetime.datetime.now()
986 # Timedrift is updated only in fractions in order to prevent problems, but we need an integer index
987 timedrift = round(timedrift)
988 secret = bytes.decode(base64.b32encode(bytes.fromhex(secret))) # decode secret
989 otp = str(otp).zfill(6) # fill with zeros in front
991 # logging.debug(f"TimeBasedOTP:verify: {digest=}, {interval=}, {valid_window=}")
992 totp = pyotp.TOTP(secret, digest=digest, interval=interval)
994 if valid_window:
995 for offset in range(timedrift - valid_window, timedrift + valid_window + 1):
996 token = str(totp.at(for_time, offset))
997 # logging.debug(f"TimeBasedOTP:verify: {offset=}, {otp=}, {token=}")
998 if hmac.compare_digest(otp, token):
999 return offset
1001 return None
1003 return 0 if hmac.compare_digest(otp, str(totp.at(for_time, timedrift))) else None
1005 # FIXME: VIUR4 rename
1006 def updateTimeDrift(self, user_key: db.Key, idx: float) -> None:
1007 """
1008 Updates the clock-drift value.
1010 The value is only changed in 1/10 steps, so that a late submit by an user doesn't skew
1011 it out of bounds. Maximum change per call is 0.3 minutes.
1013 :param user_key: For which user should the update occour
1014 :param idx: How many steps before/behind was that token
1015 """
1016 if user_skel := self._user_module.skel().read(user_key):
1017 if otp_skel := self._get_otptoken(user_skel):
1018 otp_skel.patch(
1019 {
1020 "+otp_timedrift": min(max(0.1 * idx, -0.3), 0.3)
1021 },
1022 update_relations=False,
1023 )
1026class AuthenticatorOTP(UserSecondFactorAuthentication):
1027 """
1028 This class handles the second factor for apps like authy and so on
1029 """
1030 METHOD_NAME = "X-VIUR-2FACTOR-AuthenticatorOTP"
1032 # second_factor_add_template = "user_secondfactor_add"
1033 # """Template to configure (add) a new TOPT"""
1035 ACTION_NAME = "authenticator_otp"
1036 """Action name provided for *otp_template* on login"""
1038 NAME = "Authenticator App"
1040 # FIXME: The second factor add has to be rewritten entirely to ActionSkel paradigm.
1041 '''
1042 @exposed
1043 @force_ssl
1044 @skey(allow_empty=True)
1045 def add(self, otp=None):
1046 """
1047 We try to read the otp_app_secret from the current session. When this fails we generate a new one and store
1048 it in the session.
1050 If an otp and a skey are provided we are validate the skey and the otp. If both is successfully we store
1051 the otp_app_secret from the session in the user entry.
1052 """
1053 current_session = current.session.get()
1055 if not (otp_app_secret := current_session.get("_maybe_otp_app_secret")):
1056 otp_app_secret = AuthenticatorOTP.generate_otp_app_secret()
1057 current_session["_maybe_otp_app_secret"] = otp_app_secret
1058 current_session.markChanged()
1060 if otp is None:
1061 return self._user_module.render.second_factor_add(
1062 tpl=self.second_factor_add_template,
1063 action_name=self.ACTION_NAME,
1064 name=i18n.translate(
1065 f"viur.core.modules.user.auth{self.ACTION_NAME}",
1066 default_variables={"name": self.NAME},
1067 ),
1068 add_url=self.add_url,
1069 otp_uri=AuthenticatorOTP.generate_otp_app_secret_uri(otp_app_secret))
1070 else:
1071 if not AuthenticatorOTP.verify_otp(otp, otp_app_secret):
1072 return self._user_module.render.second_factor_add(
1073 tpl=self.second_factor_add_template,
1074 action_name=self.ACTION_NAME,
1075 name=i18n.translate(
1076 f"viur.core.modules.user.auth.{self.ACTION_NAME}",
1077 default_variables={"name": self.NAME},
1078 ),
1079 add_url=self.add_url,
1080 otp_uri=AuthenticatorOTP.generate_otp_app_secret_uri(otp_app_secret)) # to add errors
1082 # Now we can set the otp_app_secret to the current User and render der Success-template
1083 AuthenticatorOTP.set_otp_app_secret(otp_app_secret)
1084 return self._user_module.render.second_factor_add_success(
1085 action_name=self.ACTION_NAME,
1086 name=i18n.translate(
1087 f"viur.core.modules.user.auth.{self.ACTION_NAME}",
1088 default_variables={"name": self.NAME},
1089 ),
1090 )
1091 '''
1093 def can_handle(self, skel: skeleton.SkeletonInstance) -> bool:
1094 """
1095 We can only handle the second factor if we have stored an otp_app_secret before.
1096 """
1097 return bool(skel.dbEntity.get("otp_app_secret", ""))
1099 @classmethod
1100 def patch_user_skel(cls, skel_cls):
1101 """
1102 Modifies the UserSkel to be equipped by bones required by Authenticator App
1103 """
1104 # Authenticator OTP Apps (like Authy)
1105 skel_cls.otp_app_secret = CredentialBone(
1106 descr="OTP Secret (App-Key)",
1107 params={
1108 "category": "Second Factor Authentication",
1109 },
1110 )
1112 @classmethod
1113 def set_otp_app_secret(cls, otp_app_secret=None):
1114 """
1115 Write a new OTP Token in the current user entry.
1116 """
1117 if otp_app_secret is None:
1118 logging.error("No 'otp_app_secret' is provided")
1119 raise errors.PreconditionFailed("No 'otp_app_secret' is provided")
1120 if not (cuser := current.user.get()):
1121 raise errors.Unauthorized()
1123 def transaction(user_key):
1124 if not (user := db.get(user_key)):
1125 raise errors.NotFound()
1126 user["otp_app_secret"] = otp_app_secret
1127 db.put(user)
1129 db.run_in_transaction(transaction, cuser["key"])
1131 @classmethod
1132 def generate_otp_app_secret_uri(cls, otp_app_secret) -> str:
1133 """
1134 :return an otp uri like otpauth://totp/Example:alice@google.com?secret=ABCDEFGH1234&issuer=Example
1135 """
1136 if not (cuser := current.user.get()):
1137 raise errors.Unauthorized()
1138 if not (issuer := conf.user.otp_issuer):
1139 logging.warning(
1140 f"conf.user.otp_issuer is None we replace the issuer by {conf.instance.project_id=}")
1141 issuer = conf.instance.project_id
1143 return pyotp.TOTP(otp_app_secret).provisioning_uri(name=cuser["name"], issuer_name=issuer)
1145 @classmethod
1146 def generate_otp_app_secret(cls) -> str:
1147 """
1148 Generate a new OTP Secret
1149 :return an otp
1150 """
1151 return pyotp.random_base32()
1153 @classmethod
1154 def verify_otp(cls, otp: str | int, secret: str) -> bool:
1155 return pyotp.TOTP(secret).verify(otp)
1157 @exposed
1158 def start(self):
1159 otp_user_conf = {"attempts": 0}
1160 session = current.session.get()
1161 session["_otp_user"] = otp_user_conf
1162 session.markChanged()
1163 return self._user_module.render.edit(
1164 TimeBasedOTP.OtpSkel(),
1165 params={
1166 "name": i18n.translate(
1167 f"viur.core.modules.user.auth.{self.ACTION_NAME}",
1168 default_variables={"name": self.NAME},
1169 ),
1170 "action_name": self.ACTION_NAME,
1171 "action_url": self.action_url,
1172 },
1173 tpl=self.second_factor_login_template,
1174 )
1176 @exposed
1177 @force_ssl
1178 @skey
1179 def authenticator_otp(self, **kwargs):
1180 """
1181 We verify the otp here with the secret we stored before.
1182 """
1183 session = current.session.get()
1184 user_key = db.Key(self._user_module.kindName, session["possible_user_key"])
1186 if not (otp_user_conf := session.get("_otp_user")): 1186 ↛ 1187line 1186 didn't jump to line 1187 because the condition on line 1186 was never true
1187 raise errors.PreconditionFailed("No OTP process started in this session")
1189 # Check if maximum second factor verification attempts
1190 if (attempts := otp_user_conf.get("attempts") or 0) > self.MAX_RETRY: 1190 ↛ 1191line 1190 didn't jump to line 1191 because the condition on line 1190 was never true
1191 raise errors.Forbidden("Maximum amount of authentication retries exceeded")
1193 if not (user := db.get(user_key)): 1193 ↛ 1194line 1193 didn't jump to line 1194 because the condition on line 1193 was never true
1194 raise errors.NotFound()
1196 skel = TimeBasedOTP.OtpSkel()
1197 if not skel.fromClient(kwargs): 1197 ↛ 1198line 1197 didn't jump to line 1198 because the condition on line 1197 was never true
1198 raise errors.PreconditionFailed()
1199 otp_token = str(skel["otptoken"]).zfill(6)
1201 if AuthenticatorOTP.verify_otp(otp=otp_token, secret=user["otp_app_secret"]): 1201 ↛ 1202line 1201 didn't jump to line 1202 because the condition on line 1201 was never true
1202 return self._user_module.secondFactorSucceeded(self, user_key)
1203 otp_user_conf["attempts"] = attempts + 1
1204 session.markChanged()
1205 skel.errors = [ReadFromClientError(ReadFromClientErrorSeverity.Invalid, "Wrong OTP Token", ["otptoken"])]
1206 return self._user_module.render.edit(
1207 skel,
1208 params={
1209 "name": i18n.translate(
1210 f"viur.core.modules.user.auth.{self.ACTION_NAME}",
1211 default_variables={"name": self.NAME},
1212 ),
1213 "action_name": self.ACTION_NAME,
1214 "action_url": self.action_url,
1215 },
1216 tpl=self.second_factor_login_template,
1217 )
1220_HOSTNAME_RE = re.compile(r"[a-z0-9.-]+")
1223def _is_redirect_allowed(url: str, whitelist: t.Iterable[str]) -> bool:
1224 """
1225 Checks whether *url* matches one of the redirect patterns in *whitelist*.
1227 Scheme, host, port and path are matched separately (via :func:`fnmatch.fnmatchcase`),
1228 so a wildcard in a pattern can never reach beyond its component, e.g. from the host
1229 into the path, the query or a user-info part of *url*. Query and fragment of *url*
1230 are not considered.
1232 URLs with user info (``user@host``), backslashes, a scheme other than http/https or
1233 a hostname containing anything but letters, digits, dots and dashes are rejected,
1234 because browsers may resolve them to a different host than :func:`urllib.parse.urlsplit`.
1235 This also rejects IPv6 literals.
1237 A pattern without a port only matches URLs without an explicit port;
1238 a pattern without a path matches any path. The pattern ``*`` allows any URL.
1239 """
1240 if "\\" in url:
1241 return False
1243 try:
1244 target = urllib.parse.urlsplit(url)
1245 target_port = target.port
1246 except ValueError:
1247 return False
1249 if (
1250 target.scheme not in ("http", "https")
1251 or "@" in target.netloc
1252 or not target.hostname
1253 or not _HOSTNAME_RE.fullmatch(target.hostname)
1254 ):
1255 return False
1257 for pattern in whitelist:
1258 if pattern == "*":
1259 return True
1261 scheme, sep, rest = pattern.partition("://")
1262 if not sep:
1263 continue
1265 netloc, slash, path = rest.partition("/")
1266 host, has_port, port = netloc.partition(":")
1268 if not fnmatch.fnmatchcase(target.scheme, scheme.lower()):
1269 continue
1271 if not fnmatch.fnmatchcase(target.hostname, host.lower()):
1272 continue
1274 if has_port:
1275 if target_port is None or not fnmatch.fnmatchcase(str(target_port), port):
1276 continue
1277 elif target_port is not None:
1278 continue
1280 if slash and not fnmatch.fnmatchcase(target.path or "/", f"/{path}"):
1281 continue
1283 return True
1285 return False
1288def _has_app_login_access() -> bool:
1289 """
1290 Checks the current user against :attr:`conf.user.app_login_access`.
1292 ``None`` grants access to every authenticated user. Otherwise the format of
1293 :func:`viur.core.decorators.access` applies: any entry grants access, and a list, tuple or
1294 set as entry requires all of its rights.
1295 """
1296 if (required := conf.user.app_login_access) is None:
1297 return True
1299 user_access = current.user.get()["access"]
1300 return any(
1301 all(right in user_access for right in utils.ensure_iterable(entry))
1302 for entry in utils.ensure_iterable(required)
1303 )
1306class User(List):
1307 """
1308 The User module is used to manage and authenticate users in a ViUR system.
1310 It is used in almost any ViUR project, but ViUR can also function without any user capabilites.
1311 """
1313 kindName = "user"
1314 addTemplate = "user_add"
1315 addSuccessTemplate = "user_add_success"
1317 authenticationProviders: t.Iterable[UserPrimaryAuthentication] = tuple(filter(
1318 None, (
1319 UserPassword,
1320 conf.user.google_client_id and GoogleAccount,
1321 )
1322 ))
1323 """
1324 Specifies primary authentication providers that are made available
1325 as sub-modules under `user/auth_<classname>`. They might require
1326 customization or configuration.
1327 """
1329 secondFactorProviders: t.Iterable[UserSecondFactorAuthentication] = (
1330 TimeBasedOTP,
1331 AuthenticatorOTP,
1332 )
1333 """
1334 Specifies secondary authentication providers that are made available
1335 as sub-modules under `user/f2_<classname>`. They might require
1336 customization or configuration, which is determined during the
1337 login-process depending on the user that wants to login.
1338 """
1340 validAuthenticationMethods = tuple(filter(
1341 None, (
1342 (UserPassword, AuthenticatorOTP),
1343 (UserPassword, TimeBasedOTP),
1344 (UserPassword, None),
1345 (GoogleAccount, None) if conf.user.google_client_id else None,
1346 )
1347 ))
1348 """
1349 Specifies the possible combinations of primary- and secondary factor
1350 login methos.
1352 GoogleLogin defaults to no second factor, as the Google Account can be
1353 secured by a secondary factor. AuthenticatorOTP and TimeBasedOTP are only
1354 handled when there is a user-dependent configuration available.
1355 """
1357 msg_missing_second_factor = "Second factor required but not configured for this user."
1359 secondFactorTimeWindow = datetime.timedelta(minutes=10)
1361 default_order = "name.idx"
1363 roles = {
1364 "admin": "*",
1365 }
1367 def __init__(self, moduleName, modulePath):
1368 for provider in self.authenticationProviders:
1369 assert issubclass(provider, UserPrimaryAuthentication)
1370 name = f"auth_{provider.__name__.lower()}"
1371 setattr(self, name, provider(name, f"{modulePath}/{name}", self))
1373 for provider in self.secondFactorProviders:
1374 assert issubclass(provider, UserSecondFactorAuthentication)
1375 name = f"f2_{provider.__name__.lower()}"
1376 setattr(self, name, provider(name, f"{modulePath}/{name}", self))
1378 super().__init__(moduleName, modulePath)
1380 def adminInfo(self):
1381 ret = {
1382 "icon": "person-fill",
1383 }
1385 if self.is_admin(current.user.get()):
1386 ret |= {
1387 "actions": [
1388 "trigger_kick",
1389 "trigger_takeover",
1390 ],
1391 "customActions": {
1392 "trigger_kick": {
1393 "name": i18n.translate(
1394 key="viur.core.modules.user.customActions.kick",
1395 defaultText="Kick user",
1396 hint="Title of the kick user function"
1397 ),
1398 "icon": "trash2-fill",
1399 "action": "fetch",
1400 "url": "/vi/{{module}}/trigger/kick/{{key}}",
1401 "confirm": i18n.translate(
1402 key="viur.core.modules.user.customActions.kick.confirm",
1403 defaultText="Do you really want to drop all sessions of the selected user from the system?",
1404 ),
1405 "success": i18n.translate(
1406 key="viur.core.modules.user.customActions.kick.success",
1407 defaultText="Sessions of the user are being invalidated.",
1408 ),
1409 },
1410 "trigger_takeover": {
1411 "name": i18n.translate(
1412 key="viur.core.modules.user.customActions.takeover",
1413 defaultText="Take-over user",
1414 hint="Title of the take user over function"
1415 ),
1416 "icon": "file-person-fill",
1417 "action": "fetch",
1418 "url": "/vi/{{module}}/trigger/takeover/{{key}}",
1419 "confirm": i18n.translate(
1420 key="viur.core.modules.user.customActions.takeover.confirm",
1421 defaultText="Do you really want to replace your current user session by a "
1422 "user session of the selected user?",
1423 ),
1424 "success": i18n.translate(
1425 key="viur.core.modules.user.customActions.takeover.success",
1426 defaultText="You're now know as the selected user!",
1427 ),
1428 "then": "reload-vi",
1429 },
1430 },
1431 }
1433 return ret
1435 def get_role_defaults(self, role: str) -> set[str]:
1436 """
1437 Returns a set of default access rights for a given role.
1439 Defaults to "admin" usage for any role > "user"
1440 and "scriptor" usage for "admin" role.
1441 """
1442 ret = set()
1444 if role in ("viewer", "editor", "admin"):
1445 ret.add("admin")
1447 if role == "admin":
1448 ret.add("scriptor")
1450 return ret
1452 def addSkel(self) -> skeleton.SkeletonInstance["UserSkel"]:
1453 skel = super().addSkel().clone()
1455 if self.is_admin(current.user.get()):
1456 # An admin tries to add a new user.
1457 skel.status.readOnly = False
1458 skel.status.visible = True
1459 skel.access.readOnly = False
1460 skel.access.visible = True
1462 else:
1463 skel.status.readOnly = True
1464 skel["status"] = Status.UNSET
1465 skel.status.visible = False
1466 skel.access.readOnly = True
1467 skel["access"] = []
1468 skel.access.visible = False
1470 if "password" in skel:
1471 # Unlock and require a password
1472 skel.password.required = True
1473 skel.password.visible = True
1474 skel.password.readOnly = False
1476 skel.name.readOnly = False # Don't enforce readonly name in user/add
1477 return skel
1479 def editSkel(self, *args, **kwargs) -> skeleton.SkeletonInstance["UserSkel"]:
1480 skel = super().editSkel().clone()
1482 if "password" in skel:
1483 skel.password.required = False
1484 skel.password.visible = True
1485 skel.password.readOnly = False
1487 lock = not self.is_admin(current.user.get())
1488 skel.name.readOnly = lock
1489 skel.access.readOnly = lock
1490 skel.status.readOnly = lock
1492 return skel
1494 def secondFactorProviderByClass(self, cls) -> UserSecondFactorAuthentication:
1495 return getattr(self, f"f2_{cls.__name__.lower()}")
1497 def getCurrentUser(self):
1498 session = current.session.get()
1500 req = current.request.get()
1501 if session and (session.loaded or req.is_deferred) and (user := session.get("user")):
1502 skel = self.baseSkel()
1503 skel.setEntity(user)
1504 return skel
1506 return None
1508 def continueAuthenticationFlow(self, provider: UserPrimaryAuthentication, user_key: db.Key):
1509 """
1510 Continue authentication flow when primary authentication succeeded.
1511 """
1512 skel = self.baseSkel()
1514 if not skel.read(user_key):
1515 raise errors.NotFound("User was not found.")
1517 if not provider.can_handle(skel):
1518 raise errors.Forbidden("User is not allowed to use this primary login method.")
1520 session = current.session.get()
1521 session["possible_user_key"] = user_key.id_or_name
1522 session["_secondFactorStart"] = utils.utcNow()
1523 session.markChanged()
1525 second_factor_providers = []
1527 for auth_provider, second_factor in self.validAuthenticationMethods:
1528 if isinstance(provider, auth_provider):
1529 if second_factor is not None:
1530 second_factor_provider_instance = self.secondFactorProviderByClass(second_factor)
1531 if second_factor_provider_instance.can_handle(skel):
1532 second_factor_providers.append(second_factor_provider_instance)
1533 else:
1534 second_factor_providers.append(None)
1536 if len(second_factor_providers) > 1 and None in second_factor_providers:
1537 # We have a second factor. So we can get rid of the None
1538 second_factor_providers.pop(second_factor_providers.index(None))
1540 if len(second_factor_providers) == 0:
1541 raise errors.NotAcceptable(self.msg_missing_second_factor)
1542 elif len(second_factor_providers) == 1:
1543 if second_factor_providers[0] is None:
1544 # We allow sign-in without a second factor
1545 return self.authenticateUser(user_key)
1546 # We have only one second factor we don't need the choice template
1547 return second_factor_providers[0].start(user_key)
1549 # In case there is more than one second factor provider remaining, let the user decide!
1550 current.session.get()["_secondfactor_providers"] = {
1551 second_factor.start_url: second_factor.NAME
1552 for second_factor in second_factor_providers
1553 if second_factor.VISIBLE
1554 }
1556 return self.select_secondfactor_provider()
1558 def secondFactorSucceeded(self, provider: UserSecondFactorAuthentication, user_key: db.Key):
1559 """
1560 Continue authentication flow when secondary authentication succeeded.
1561 """
1562 session = current.session.get()
1563 if session["possible_user_key"] != user_key.id_or_name:
1564 raise errors.Forbidden()
1566 # Assert that the second factor verification finished in time
1567 if utils.utcNow() - session["_secondFactorStart"] > self.secondFactorTimeWindow:
1568 raise errors.RequestTimeout()
1570 return self.authenticateUser(user_key)
1572 def is_active(self, skel: skeleton.SkeletonInstance) -> bool | None:
1573 """
1574 Hookable check if a user is defined as "active" and can login.
1576 :param skel: The UserSkel of the user who wants to login.
1577 :returns: Returns True or False when the result is unambigous and the user is active or not. \
1578 Returns None when the provided skel doesn't provide enough information for determination.
1579 """
1580 if skel and "status" in skel:
1581 status = skel["status"]
1582 if not isinstance(status, (Status, int)):
1583 try:
1584 status = int(status)
1585 except ValueError:
1586 status = Status.UNSET
1588 return status >= Status.ACTIVE
1590 return None
1592 def is_admin(self, skel: skeleton.SkeletonInstance) -> bool | None:
1593 """
1594 Hookable check if a user is defined as "admin" and can edit or log into other users.
1595 Defaults to "root" users only.
1597 :param skel: The UserSkel of the user who wants should be checked for user admin privileges.
1598 :returns: Returns True or False when the result is unambigous and the user is admin or not. \
1599 Returns None when the provided skel doesn't provide enough information for determination.
1600 """
1601 if skel and "access" in skel:
1602 return "root" in skel["access"]
1604 return None
1606 def authenticateUser(self, key: db.Key, **kwargs):
1607 """
1608 Performs Log-In for the current session and the given user key.
1610 This resets the current session: All fields not explicitly marked as persistent
1611 by conf.user.session_persistent_fields_on_login are gone afterwards.
1613 :param key: The (DB-)Key of the user we shall authenticate
1614 """
1615 skel = self.baseSkel()
1616 if not skel.read(key):
1617 raise ValueError(f"Unable to authenticate unknown user {key}")
1619 # Verify that this user account is active
1620 if not self.is_active(skel):
1621 raise errors.Forbidden("The user is disabled and cannot be authenticated.")
1623 # Update session for user
1624 session = current.session.get()
1625 # Remember persistent fields...
1626 take_over = {k: v for k, v in session.items() if k in conf.user.session_persistent_fields_on_login}
1627 session.reset()
1628 # and copy them over to the new session
1629 session |= take_over
1631 # Update session, user and request
1632 session["user"] = skel.dbEntity
1634 current.request.get().response.headers[securitykey.SECURITYKEY_STATIC_HEADER] = session.static_security_key
1635 current.user.set(self.getCurrentUser())
1637 self.onLogin(skel)
1639 return self.render.render("login_success", skel, **kwargs)
1642 # Action for primary authentication selection
1644 def SelectAuthenticationProviderSkel(self):
1645 providers = {}
1646 first = None
1647 for provider in self.authenticationProviders:
1648 if not provider.VISIBLE:
1649 continue
1651 provider = getattr(self, f"auth_{provider.__name__.lower()}")
1652 providers[provider.start_url] = provider.NAME
1654 if first is None:
1655 first = provider.start_url
1657 class SelectAuthenticationProviderSkel(skeleton.RelSkel):
1658 provider = SelectBone(
1659 descr="Authentication method",
1660 required=True,
1661 values=providers,
1662 defaultValue=first,
1663 )
1665 return SelectAuthenticationProviderSkel()
1667 @exposed
1668 def select_authentication_provider(self, **kwargs):
1669 skel = self.SelectAuthenticationProviderSkel()
1671 # Read required bones from client
1672 if len(skel.provider.values) > 1 and (not kwargs or not skel.fromClient(kwargs)):
1673 return self.render.render("select_authentication_provider", skel)
1675 return self.render.render("select_authentication_provider_success", skel, next_url=skel["provider"])
1677 # Action for second factor select
1679 class SelectSecondFactorProviderSkel(skeleton.RelSkel):
1680 provider = SelectBone(
1681 descr="Second factor",
1682 required=True,
1683 values=lambda: current.session.get()["_secondfactor_providers"] or (),
1684 )
1686 @exposed
1687 def select_secondfactor_provider(self, **kwargs):
1688 skel = self.SelectSecondFactorProviderSkel()
1690 # Read required bones from client
1691 if not kwargs or not skel.fromClient(kwargs):
1692 return self.render.render("select_secondfactor_provider", skel)
1694 del current.session.get()["_secondfactor_providers"]
1696 return self.render.render("select_secondfactor_provider_success", skel, next_url=skel["provider"])
1698 @exposed
1699 @skey
1700 def logout(self, **kwargs):
1701 """
1702 Implements the logout action. It also terminates the current session (all keys not listed
1703 in viur.session_persistent_fields_on_logout will be lost).
1704 """
1705 if not (user := current.user.get()):
1706 raise errors.Unauthorized()
1708 self.onLogout(user)
1710 session = current.session.get()
1712 if take_over := {k: v for k, v in session.items() if k in conf.user.session_persistent_fields_on_logout}:
1713 session.reset()
1714 session |= take_over
1715 else:
1716 session.clear()
1718 current.user.set(None) # set user to none in context var
1720 return self.render.render("logout_success")
1722 @exposed
1723 def login(self, *args, **kwargs):
1724 return self.select_authentication_provider()
1726 def onLogin(self, skel: skeleton.SkeletonInstance):
1727 """
1728 Hook to be called on user login.
1729 """
1730 # Update the lastlogin timestamp (if available!)
1731 if "lastlogin" in skel:
1732 now = utils.utcNow()
1734 # Conserve DB-Writes: Update the user max once in 30 Minutes (why??)
1735 if not skel["lastlogin"] or ((now - skel["lastlogin"]) > datetime.timedelta(minutes=30)):
1736 skel["lastlogin"] = now
1737 skel.write(update_relations=False)
1739 logging.info(f"""User {skel["name"]} logged in""")
1741 def onLogout(self, skel: skeleton.SkeletonInstance):
1742 """
1743 Hook to be called on user logout.
1744 """
1745 logging.info(f"""User {skel["name"]} logged out""")
1747 @exposed
1748 def view(self, key: db.KeyType = "self", *args, **kwargs):
1749 """
1750 Allow a special key "self" to reference the current user.
1752 By default, any authenticated user can view its own user entry,
1753 to obtain access rights and any specific user information.
1754 This behavior is defined in the customized `canView` function,
1755 which is overwritten by the User-module.
1757 The rendered skeleton can be modified or restriced by specifying
1758 a customized view-skeleton.
1759 """
1760 if key == "self":
1761 if user := current.user.get():
1762 key = user["key"]
1763 else:
1764 raise errors.Unauthorized("Cannot view 'self' with unknown user")
1766 return super().view(key, *args, **kwargs)
1768 def canView(self, skel) -> bool:
1769 if user := current.user.get():
1770 if skel["key"] == user["key"]:
1771 return True
1773 if self.is_admin(user) or "user-view" in user["access"]:
1774 return True
1776 return False
1778 @exposed
1779 @skey(allow_empty=True)
1780 def edit(self, key: db.KeyType = "self", *args, **kwargs):
1781 """
1782 Allow a special key "self" to reference the current user.
1784 This modification will only allow to use "self" as a key;
1785 The specific access right to let the user edit itself must
1786 still be customized.
1788 The rendered and editable skeleton can be modified or restriced
1789 by specifying a customized edit-skeleton.
1790 """
1791 if key == "self":
1792 if user := current.user.get():
1793 key = user["key"]
1794 else:
1795 raise errors.Unauthorized("Cannot edit 'self' with unknown user")
1797 return super().edit(key, *args, **kwargs)
1799 @exposed
1800 def getAuthMethods(self, *args, **kwargs):
1801 """Legacy method prior < viur-core 3.8: Inform tools like Admin which authentication to use"""
1802 logging.warning("DEPRECATED!!! Use '/user/login'-method for this, or update your admin version!")
1804 res = [
1805 (primary.METHOD_NAME, secondary.METHOD_NAME if secondary else None)
1806 for primary, secondary in self.validAuthenticationMethods
1807 ]
1809 return json.dumps(res)
1811 @exposed
1812 def trigger(self, action: str, key: str):
1813 # Check for provided access right definition (equivalent to client-side check), fallback to root!
1814 access = self.adminInfo().get("customActions", {}).get(f"trigger_{action}", {}).get("access") or ()
1815 if not (
1816 (cuser := current.user.get())
1817 and (
1818 any(role in cuser["access"] for role in access)
1819 or self.is_admin(cuser)
1820 )
1821 ):
1822 raise errors.Unauthorized()
1824 skel = self.skel()
1825 if not skel.read(key) and not (skel := skel.all().mergeExternalFilter({"name": key}).getSkel()):
1826 raise errors.NotFound("The provided user does not exist.")
1828 match action:
1829 case "takeover":
1830 self.authenticateUser(skel["key"])
1832 case "kick":
1833 session.killSessionByUser(skel["key"])
1835 case _:
1836 raise errors.NotImplemented(f"Action {action!r} not implemented")
1838 return self.render.render(f"trigger/{action}Success", skel)
1840 @exposed
1841 @access(_has_app_login_access, offer_login=True)
1842 def get_cookie_for_app(self, redirect_to: str = None):
1843 """
1844 Generates a session cookie for the currently logged-in user and hands it to an external
1845 client (script, native app, or WebView).
1847 This endpoint is the entry point of a *App Login Flow*. A logged-in user
1848 authenticates normally in the browser and then opens this URL. Which users may do so
1849 is configured by :attr:`conf.user.app_login_access` (default: every authenticated user).
1850 The backend creates a fresh ViUR session (see :meth:`_get_cookie_for_app`) and
1851 delivers the resulting ``Set-Cookie`` string to the caller.
1853 **Typical usage — local Python client / script:**
1855 The caller spins up a temporary local HTTP server (e.g. on ``http://localhost:60000``)
1856 and passes its address as *redirect_to*::
1858 /vi/user/get_cookie_for_app?redirect_to=http://localhost:60000
1860 After the user authenticates in the browser, the backend redirects to::
1862 http://localhost:60000?cookie=<url-encoded Set-Cookie string>&app=<project_id>
1864 The local server can then extract only the ``name=value`` part of the cookie string
1865 (everything before the first ``;``) and use it for subsequent API calls::
1867 cookie_str = qs["cookie"][0] # full Set-Cookie value
1868 key, value = cookie_str.split(";", 1)[0].split("=")
1869 session.cookies.update({key: value})
1871 The ``app`` query parameter is set by the server to ``conf.instance.project_id`` and
1872 lets the client distinguish between multiple backends / cache credentials per project.
1874 **Alternative — WebView / browser redirect:**
1876 When the receiving side is a browser or WebView the full ``Set-Cookie`` string can be
1877 forwarded to :meth:`apply_login_cookie` to let the framework activate the session.
1879 :param redirect_to: Optional callback URL. When provided the caller is redirected to
1880 that URL with two query parameters appended automatically:
1882 - ``cookie`` – URL-encoded ``Set-Cookie`` string (``name=value;flags…``).
1883 - ``app`` – the server's GCP project ID (``conf.instance.project_id``).
1885 A ``?`` is appended if the URL does not already contain one.
1886 When omitted, the raw ``Set-Cookie`` string is returned as ``text/plain``
1887 (useful for debugging or direct API calls).
1889 .. warning:: **Open-redirect / session-hijacking risk**
1891 Because the session cookie is appended as a plain query parameter, an
1892 attacker who can convince an authenticated user to click a crafted link
1893 (e.g. via phishing) could redirect the browser to an evil server that
1894 simply harvests the ``cookie`` parameter and gains full session access.
1896 To mitigate this, all ``redirect_to`` values are validated against
1897 :attr:`conf.user.redirect_whitelist`. Scheme, host, port and path are matched
1898 separately, so a wildcard cannot reach from the host into the path or query.
1899 Only explicitly whitelisted URL patterns are accepted;
1900 anything else is rejected with ``403 Forbidden``.
1901 Configure the whitelist in your project to include every legitimate
1902 callback origin (local scripts, internal tooling, etc.).
1904 :raises errors.Forbidden: When *redirect_to* does not match any pattern in
1905 :attr:`conf.user.redirect_whitelist`.
1906 :raises errors.Redirect: Always raised when *redirect_to* is supplied and allowed.
1907 """
1908 if redirect_to:
1909 whitelist = utils.ensure_iterable(conf.user.redirect_whitelist)
1910 if not _is_redirect_allowed(redirect_to, whitelist):
1911 raise errors.Forbidden(f"Redirect target is not whitelisted")
1912 if "?" not in redirect_to:
1913 redirect_to = f"{redirect_to}?"
1914 raise errors.Redirect(
1915 f"{redirect_to}"
1916 f"&cookie={urllib.parse.quote_plus(self._get_cookie_for_app())}"
1917 f"&app={conf.instance.project_id}"
1918 )
1919 current.request.get().response.headers["Content-Type"] = "text/plain"
1920 return self._get_cookie_for_app()
1922 def _get_cookie_for_app(self) -> str:
1923 """
1924 Creates a new, standalone ViUR session for the current user and returns the
1925 corresponding ``Set-Cookie`` header value.
1927 Unlike the regular session created during a normal login, this session is intentionally
1928 **not** attached to the current HTTP request. Instead it is persisted directly in
1929 Datastore so that a different HTTP client can pick it up via :meth:`apply_login_cookie`.
1931 The created session entity mirrors the structure of a regular :class:`Session` entry:
1933 - ``data["user"]`` – the full user ``dbEntity`` (needed by the session loader).
1934 - ``data["is_app_session"]``– flag to distinguish app sessions from regular browser sessions.
1935 - ``static_security_key`` – random value, same role as in normal sessions.
1936 - ``lastseen`` – current timestamp so the session is not immediately garbage-
1937 collected.
1938 - ``user`` – stringified user key for server-side user-based queries.
1940 :returns: A ``Set-Cookie`` header value in the form
1941 ``<cookie_name>=<key>;<flags>`` where ``<flags>`` is produced by
1942 :meth:`Session.build_flags` (``Path=/; HttpOnly; SameSite=…; Secure; Max-Age=…``).
1943 """
1944 cookie_key = utils.string.random(42)
1945 db_session = db.Entity(db.Key(Session.kindName, cookie_key))
1946 data = db.Entity()
1947 data["user"] = current.user.get().dbEntity
1948 data["is_app_session"] = True
1949 db_session["data"] = db.fix_unindexable_properties(data)
1950 db_session["static_security_key"] = utils.string.random(42)
1951 db_session["lastseen"] = time.time()
1952 db_session["user"] = str(current.user.get()["key"])
1953 db_session.exclude_from_indexes = {"data"}
1954 db.put(db_session)
1956 # Provide Set-Cookie header entry with configured properties
1957 return f"{Session.cookie_name}={cookie_key};{Session.build_flags()}"
1959 @exposed
1960 def apply_login_cookie(self, cookie: str):
1961 """
1962 Redirect endpoint to load session from the given cookie.
1964 This is the second half of the *App Login Flow*. A native app or WebView that received a
1965 ``Set-Cookie`` string from :meth:`get_cookie_for_app` (typically via a redirect URL
1966 parameter) calls this endpoint to activate the embedded session for its own HTTP context.
1968 The flow is:
1970 1. Parse the raw ``Set-Cookie`` string with :class:`http.cookies.SimpleCookie`.
1971 2. Look for the expected session cookie name (:attr:`Session.cookie_name`).
1972 3. Reset the caller's current (anonymous) session.
1973 4. Inject the cookie value into the current request's cookie jar so that
1974 :meth:`Session.load` can find the pre-built Datastore session.
1975 5. Redirect to ``/`` – from this point on the caller is fully authenticated.
1977 :param cookie: A raw ``Set-Cookie`` header value as produced by :meth:`_get_cookie_for_app`,
1978 e.g. ``viur_cookie_myproject=<key>;Path=/;HttpOnly;…``.
1979 :raises errors.Redirect: On success – redirects to ``/``.
1980 :raises errors.BadRequest: When the cookie string does not contain a recognisable session
1981 cookie (i.e. :attr:`Session.cookie_name` is absent after parsing).
1982 """
1983 cookies = SimpleCookie()
1984 cookies.load(cookie)
1985 if Session.cookie_name in cookies:
1986 session_cookie = cookies[Session.cookie_name]
1987 current.session.get().reset()
1988 current.request.get().request.cookies[session_cookie.key] = session_cookie.value
1989 current.session.get().load()
1990 raise errors.Redirect("/")
1991 else:
1992 raise errors.BadRequest
1994 def onEdited(self, skel):
1995 super().onEdited(skel)
1997 # In case the user is set to inactive, kill all sessions
1998 if self.is_active(skel) is False:
1999 session.killSessionByUser(skel["key"])
2001 # Otherwise, update the user entity cached in all the user's sessions
2002 else:
2003 session.update_session_user(skel["key"])
2005 def onDeleted(self, skel):
2006 super().onDeleted(skel)
2007 # Invalidate all sessions of that user
2008 session.killSessionByUser(skel["key"])
2011@tasks.StartupTask
2012def createNewUserIfNotExists():
2013 """
2014 Create a new Admin user, if the userDB is empty
2015 """
2016 if (
2017 (user_module := getattr(conf.main_app.vi, "user", None))
2018 and isinstance(user_module, User)
2019 and "addSkel" in dir(user_module)
2020 and "validAuthenticationMethods" in dir(user_module)
2021 # UserPassword must be one of the primary login methods
2022 and any(
2023 issubclass(provider[0], UserPassword)
2024 for provider in user_module.validAuthenticationMethods
2025 )
2026 ):
2027 if not db.Query(user_module.addSkel().kindName).getEntry(): # There's currently no user in the database
2028 addSkel = skeleton.skeletonByKind(user_module.addSkel().kindName)() # Ensure we have the full skeleton
2029 uname = f"""admin@{conf.instance.project_id}.appspot.com"""
2030 pw = utils.string.random(13)
2031 addSkel["name"] = uname
2032 addSkel["status"] = Status.ACTIVE # Ensure it's enabled right away
2033 addSkel["access"] = ["root"]
2034 addSkel["password"] = pw
2036 try:
2037 addSkel.write()
2038 except Exception as e:
2039 logging.critical(f"Something went wrong when trying to add admin user {uname!r} with Password {pw!r}")
2040 logging.exception(e)
2041 return
2043 msg = f"ViUR created a new admin-user for you!\nUsername: {uname}\nPassword: {pw}"
2045 logging.warning(msg)
2046 email.send_email_to_admins("New ViUR password", msg)
2049# DEPRECATED ATTRIBUTES HANDLING
2051def __getattr__(attr):
2052 match attr:
2053 case "userSkel": 2053 ↛ 2054line 2053 didn't jump to line 2054 because the pattern on line 2053 never matched
2054 msg = f"Use of `userSkel` is deprecated; Please use `UserSkel` instead!"
2055 warnings.warn(msg, DeprecationWarning, stacklevel=2)
2056 logging.warning(msg)
2057 return UserSkel
2059 return super(__import__(__name__).__class__).__getattr__(attr)