Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/modules/user.py: 27%
762 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-25 15:02 +0000
1import abc
2import datetime
3import enum
4import fnmatch
5import hashlib
6import hmac
7import json
8import logging
9import secrets
10import time
11import urllib.parse
12import warnings
13from http.cookies import SimpleCookie
15import user_agents
17import pyotp
18import base64
19import dataclasses
20import typing as t
21from google.auth.transport import requests
22from google.oauth2 import id_token
24from viur.core import (
25 conf, current, db, email, errors, i18n,
26 securitykey, session, skeleton, tasks, utils, Module
27)
28from viur.core.decorators import *
29from viur.core.bones import *
30from viur.core.bones.password import PBKDF2_DEFAULT_ITERATIONS, encode_password
31from viur.core.prototypes.list import List
32from viur.core.ratelimit import RateLimit
33from viur.core.securityheaders import extendCsp
34from viur.core.session import Session
37class Status(enum.IntEnum):
38 """Status enum for a user
40 This is an IntEnum, so it is comparable with plain ints as well as with
41 other IntEnum classes of the same values, e.g. when a project defines its
42 own Status enum to add custom status values to a subclassed UserSkel:
44 class Status(enum.IntEnum):
45 UNSET = 0
46 WAITING_FOR_EMAIL_VERIFICATION = 1
47 WAITING_FOR_ADMIN_VERIFICATION = 2
48 DISABLED = 5
49 ACTIVE = 10
50 PENDING_REVIEW = 15 # custom, project-specific status
52 class UserSkel(user.UserSkel):
53 status = SelectBone(
54 ...,
55 values=Status,
56 defaultValue=Status.ACTIVE,
57 )
58 """
60 UNSET = 0 # Status is unset
61 WAITING_FOR_EMAIL_VERIFICATION = 1 # Waiting for email verification
62 WAITING_FOR_ADMIN_VERIFICATION = 2 # Waiting for verification through admin
63 DISABLED = 5 # Account disabled
64 ACTIVE = 10 # Active
67class UserSkel(skeleton.Skeleton):
68 kindName = "user" # this assignment is required, as this Skeleton is defined in viur-core (see #604)
70 name = EmailBone(
71 descr="E-Mail",
72 required=True,
73 readOnly=True,
74 caseSensitive=False,
75 searchable=True,
76 unique=UniqueValue(UniqueLockMethod.SameValue, True, "Username already taken"),
77 tags=("personal", "identifier", "contact"),
78 )
80 firstname = StringBone(
81 descr="Firstname",
82 searchable=True,
83 tags="personal",
84 )
86 lastname = StringBone(
87 descr="Lastname",
88 searchable=True,
89 tags="personal",
90 )
92 roles = SelectBone(
93 descr=i18n.translate("viur.core.modules.user.bone.roles", defaultText="Roles"),
94 values=conf.user.roles,
95 required=True,
96 multiple=True,
97 # fixme: This is generally broken in VIUR! See #776 for details.
98 # vfunc=lambda values:
99 # i18n.translate(
100 # "user.bone.roles.invalid",
101 # defaultText="Invalid role setting: 'custom' can only be set alone.")
102 # if "custom" in values and len(values) > 1 else None,
103 defaultValue=list(conf.user.roles.keys())[:1],
104 )
106 access = SelectBone(
107 descr=i18n.translate("viur.core.modules.user.bone.access", defaultText="Access rights"),
108 type_suffix="access",
109 values=lambda: {
110 right: i18n.translate(f"viur.core.modules.user.accessright.{right}", defaultText=right)
111 for right in sorted(conf.user.access_rights)
112 },
113 multiple=True,
114 params={
115 "readonlyIf": "'custom' not in roles" # if "custom" is not in roles, "access" is managed by the role system
116 },
117 )
119 status = SelectBone(
120 descr=i18n.translate("viur.core.modules.user.bone.status", "Account status"),
121 values=Status,
122 translation_key_prefix="viur.core.user.status.",
123 defaultValue=Status.ACTIVE,
124 required=True,
125 )
127 lastlogin = DateBone(
128 descr="Last Login",
129 readOnly=True,
130 tags=("personal", "technical"),
131 )
133 admin_config = JsonBone( # This bone stores settings from the admin
134 descr="Config for the User",
135 visible=False,
136 )
138 def __new__(cls, *args, **kwargs):
139 """
140 Constructor for the UserSkel-class, with the capability
141 to dynamically add bones required for the configured
142 authentication methods.
143 """
144 for provider in conf.main_app.vi.user.authenticationProviders:
145 assert issubclass(provider, UserPrimaryAuthentication)
146 provider.patch_user_skel(cls)
148 for provider in conf.main_app.vi.user.secondFactorProviders:
149 assert issubclass(provider, UserSecondFactorAuthentication)
150 provider.patch_user_skel(cls)
152 cls.__boneMap__ = skeleton.MetaBaseSkel.generate_bonemap(cls)
153 return super().__new__(cls, *args, **kwargs)
155 @classmethod
156 def write(cls, skel, *args, **kwargs):
157 # Roles
158 if skel["roles"] and "custom" not in skel["roles"]:
159 # Collect access rights through rules
160 access = set()
162 for role in skel["roles"]:
163 # Get default access for this role
164 access |= conf.main_app.vi.user.get_role_defaults(role)
166 # Go through all modules and evaluate available role-settings
167 for name in dir(conf.main_app.vi):
168 if name.startswith("_"):
169 continue
171 module = getattr(conf.main_app.vi, name)
172 if not isinstance(module, Module):
173 continue
175 roles = getattr(module, "roles", None) or {}
176 rights = roles.get(role, roles.get("*", ()))
178 # Convert role into tuple if it's not
179 if not isinstance(rights, (tuple, list)):
180 rights = (rights, )
182 if "*" in rights:
183 for right in module.accessRights:
184 access.add(f"{name}-{right}")
185 else:
186 for right in rights:
187 if right in module.accessRights:
188 access.add(f"{name}-{right}")
190 # special case: "edit" and "delete" actions require "view" as well!
191 if right in ("edit", "delete") and "view" in module.accessRights:
192 access.add(f"{name}-view")
194 skel["access"] = list(access)
196 return super().write(skel, *args, **kwargs)
199class UserAuthentication(Module, abc.ABC):
200 @property
201 @abc.abstractstaticmethod
202 def METHOD_NAME() -> str:
203 """
204 Define a unique method name for this authentication.
205 """
206 ...
208 @property
209 @abc.abstractstaticmethod
210 def NAME() -> str:
211 """
212 Define a descriptive name for this authentication.
213 """
214 ...
216 @property
217 @staticmethod
218 def VISIBLE(cls) -> bool:
219 """
220 Defines if the authentication method is visible to the user.
221 """
222 return True
224 def __init__(self, moduleName, modulePath, userModule):
225 super().__init__(moduleName, modulePath)
226 self._user_module = userModule
227 self.start_url = f"{self.modulePath}/login"
229 def can_handle(self, skel: skeleton.SkeletonInstance) -> bool:
230 return True
232 @classmethod
233 def patch_user_skel(cls, skel_cls: skeleton.Skeleton):
234 """
235 Allows for an UserAuthentication to patch the UserSkel
236 class with additional bones which are required for
237 the implemented authentication method.
238 """
239 ...
242class UserPrimaryAuthentication(UserAuthentication, abc.ABC):
243 """Abstract class for all primary authentication methods."""
244 registrationEnabled = False
246 @abc.abstractmethod
247 def login(self, *args, **kwargs):
248 ...
250 def next_or_finish(self, skel: skeleton.SkeletonInstance):
251 """
252 Hook that is called whenever a part of the authentication was successful.
253 It allows to perform further steps in custom authentications,
254 e.g. change a password after first login.
255 """
256 return self._user_module.continueAuthenticationFlow(self, skel["key"])
259class UserPassword(UserPrimaryAuthentication):
260 METHOD_NAME = "X-VIUR-AUTH-User-Password"
261 NAME = "Username & Password"
263 registrationEmailVerificationRequired = True
264 registrationAdminVerificationRequired = True
266 verifySuccessTemplate = "user_verify_success"
267 verifyEmailAddressMail = "user_verify_address"
268 verifyFailedTemplate = "user_verify_failed"
269 passwordRecoveryMail = "user_password_recovery"
270 passwordRecoverySuccessTemplate = "user_passwordrecover_success"
271 passwordRecoveryTemplate = "user_passwordrecover"
273 # The default rate-limit for password recovery (10 tries each 15 minutes)
274 passwordRecoveryRateLimit = RateLimit("user.passwordrecovery", 10, 15, "ip")
276 # Limit (invalid) login-retries to once per 5 seconds
277 loginRateLimit = RateLimit("user.login", 12, 1, "ip")
279 @classmethod
280 def patch_user_skel(cls, skel_cls):
281 """
282 Modifies the UserSkel to be equipped by a PasswordBone.
283 """
284 skel_cls.password = PasswordBone(
285 readOnly=True,
286 visible=False,
287 params={
288 "category": "Authentication",
289 },
290 )
292 class LoginSkel(skeleton.RelSkel):
293 name = EmailBone(
294 descr="E-Mail",
295 required=True,
296 caseSensitive=False,
297 )
298 password = PasswordBone(
299 required=True,
300 test_threshold=0,
301 tests=(),
302 raw=True,
303 )
305 class LostPasswordStep1Skel(skeleton.RelSkel):
306 name = EmailBone(
307 descr="E-Mail",
308 required=True,
309 )
311 class LostPasswordStep2Skel(skeleton.RelSkel):
312 recovery_key = StringBone(
313 descr="Recovery Key",
314 required=True,
315 params={
316 "tooltip": i18n.translate(
317 key="viur.core.modules.user.userpassword.lostpasswordstep2.recoverykey",
318 defaultText="Please enter the validation key you've received via e-mail.",
319 hint="Shown when the user needs more than 15 minutes to paste the key",
320 ),
321 }
322 )
324 class LostPasswordStep3Skel(skeleton.RelSkel):
325 # send the recovery key again, in case the password is rejected by some reason.
326 recovery_key = StringBone(
327 descr="Recovery Key",
328 visible=False,
329 )
331 password = PasswordBone(
332 descr="New Password",
333 required=True,
334 params={
335 "tooltip": i18n.translate(
336 key="viur.core.modules.user.userpassword.lostpasswordstep3.password",
337 defaultText="Please enter a new password for your account.",
338 ),
339 }
340 )
342 @exposed
343 @force_ssl
344 @skey(allow_empty=True)
345 def login(self, **kwargs):
346 # Obtain a fresh login skel
347 skel = self.LoginSkel()
349 # Read required bones from client
350 if not (kwargs and skel.fromClient(kwargs)):
351 return self._user_module.render.render("login", skel)
353 self.loginRateLimit.assertQuotaIsAvailable()
355 # query for the username. The query might find another user, but the name is being checked for equality below
356 name = skel["name"].lower().strip()
357 user_skel = self._user_module.baseSkel()
358 user_skel = user_skel.all().filter("name.idx >=", name).getSkel() or user_skel
360 # extract password hash from raw database entity (skeleton access blocks it)
361 password_data = (user_skel.dbEntity and user_skel.dbEntity.get("password")) or {}
362 iterations = password_data.get("iterations", 1001) # remember iterations; old password hashes used 1001
363 password_hash = encode_password(skel["password"], password_data.get("salt", "-invalid-"), iterations)["pwhash"]
365 # now check if the username matches
366 is_okay = secrets.compare_digest((user_skel["name"] or "").lower().strip().encode(), name.encode())
368 # next, check if the password hash matches
369 is_okay &= secrets.compare_digest(password_data.get("pwhash", b"-invalid-"), password_hash)
371 if not is_okay:
372 # Set error to all required fields
373 for name, bone in skel.items():
374 if bone.required:
375 skel.errors.append(
376 ReadFromClientError(
377 ReadFromClientErrorSeverity.Invalid,
378 i18n.translate(
379 key="viur.core.modules.user.userpassword.login.failed",
380 defaultText="Invalid username or password provided",
381 ),
382 name,
383 )
384 )
386 self.loginRateLimit.decrementQuota() # Only failed login attempts will count to the quota
387 return self._user_module.render.render("login", skel)
389 # check if iterations are below current security standards, and update if necessary.
390 if iterations < PBKDF2_DEFAULT_ITERATIONS:
391 logging.info(f"Update password hash for user {name}.")
392 # re-hash the password with more iterations
393 # FIXME: This must be done within a transaction!
394 user_skel["password"] = kwargs["password"] # will be hashed on serialize
395 user_skel.write(update_relations=False)
397 return self.next_or_finish(user_skel)
399 @exposed
400 def pwrecover(self, recovery_key: str | None = None, skey: str | None = None, **kwargs):
401 """
402 This implements a password recovery process which lets users set a new password for their account,
403 after validating a recovery key sent by email.
405 The process is as following:
407 - The user enters the registered email adress (not validated here)
408 - A random code is generated and stored as a security-ke, then sendUserPasswordRecoveryCode is called.
409 - sendUserPasswordRecoveryCode will run in the background, check if we have a user with that name
410 and send a link with the code. It runs as a deferred task so no information if a user account exists
411 is being leaked.
412 - If the user received an email, the link can be clicked to set a new password for the account.
414 To prevent automated attacks, the first step is guarded by limited calls to this function to 10 actions
415 per 15 minutes. (One complete recovery process consists of two calls).
416 """
417 self.passwordRecoveryRateLimit.assertQuotaIsAvailable()
418 current_request = current.request.get()
420 if recovery_key is None:
421 # This is the first step, where we ask for the username of the account we'll going to reset the password on
422 skel = self.LostPasswordStep1Skel()
424 if (
425 not kwargs
426 or not current_request.isPostRequest
427 or not skel.fromClient(kwargs)
428 ):
429 return self._user_module.render.render(
430 "pwrecover", skel,
431 tpl=self.passwordRecoveryTemplate,
432 )
434 # validate security key
435 if not securitykey.validate(skey):
436 raise errors.PreconditionFailed()
438 self.passwordRecoveryRateLimit.decrementQuota()
440 recovery_key = securitykey.create(
441 duration=datetime.timedelta(minutes=15),
442 key_length=conf.security.password_recovery_key_length,
443 user_name=skel["name"].lower(),
444 session_bound=False,
445 )
447 # Send the code in background
448 self.sendUserPasswordRecoveryCode(
449 skel["name"], recovery_key, current_request.request.headers["User-Agent"]
450 )
452 # step 2 is only an action-skel, and can be ignored by a direct link in the
453 # e-mail previously sent. It depends on the implementation of the specific project.
454 return self._user_module.render.render(
455 "pwrecover", self.LostPasswordStep2Skel(),
456 tpl=self.passwordRecoveryTemplate,
457 )
459 # in step 3
460 skel = self.LostPasswordStep3Skel()
462 # reset the recovery key again, in case the fromClient() fails.
463 skel["recovery_key"] = str(recovery_key).strip()
465 # check for any input; Render input-form again when incomplete.
466 if (
467 not kwargs
468 or not current_request.isPostRequest
469 or not skel.fromClient(kwargs, ignore=("recovery_key",))
470 ):
471 return self._user_module.render.render(
472 "pwrecover", skel,
473 tpl=self.passwordRecoveryTemplate,
474 )
476 # validate security key
477 if not securitykey.validate(skey):
478 raise errors.PreconditionFailed()
480 if not (recovery_request := securitykey.validate(recovery_key, session_bound=False)):
481 raise errors.PreconditionFailed(
482 i18n.translate(
483 key="viur.core.modules.user.passwordrecovery.keyexpired",
484 defaultText="The recovery key is expired or invalid. Please start the recovery process again.",
485 hint="Shown when the user needs more than 15 minutes to paste the key, or entered an invalid key."
486 )
487 )
489 self.passwordRecoveryRateLimit.decrementQuota()
491 # If we made it here, the key was correct, so we'd hopefully have a valid user for this
492 user_skel = self._user_module.viewSkel().all().filter("name.idx =", recovery_request["user_name"]).getSkel()
494 if not user_skel:
495 raise errors.NotFound(
496 i18n.translate(
497 key="viur.core.modules.user.passwordrecovery.usernotfound",
498 defaultText="There is no account with this name",
499 hint="We cant find an account with that name (Should never happen)"
500 )
501 )
503 # If the account is locked or not yet validated, abort the process.
504 if not self._user_module.is_active(user_skel):
505 raise errors.NotFound(
506 i18n.translate(
507 key="viur.core.modules.user.passwordrecovery.accountlocked",
508 defaultText="This account is currently locked. You cannot change its password.",
509 hint="Attempted password recovery on a locked account"
510 )
511 )
513 # Update the password, save the user, reset his session and show the success-template
514 user_skel["password"] = skel["password"]
515 user_skel.write(update_relations=False)
517 return self._user_module.render.render(
518 "pwrecover_success",
519 next_url=self.start_url,
520 tpl=self.passwordRecoverySuccessTemplate
521 )
523 @tasks.CallDeferred
524 def sendUserPasswordRecoveryCode(self, user_name: str, recovery_key: str, user_agent: str) -> None:
525 """
526 Sends the given recovery code to the user given in userName. This function runs deferred
527 so there's no timing sidechannel that leaks if this user exists. Per default, we'll send the
528 code by email (assuming we have working email delivery), but this can be overridden to send it
529 by SMS or other means. We'll also update the changedate for this user, so no more than one code
530 can be send to any given user in four hours.
531 """
532 if user_skel := self._user_module.viewSkel().all().filter("name.idx =", user_name).getSkel():
533 user_agent = user_agents.parse(user_agent)
534 email.send_email(
535 tpl=self.passwordRecoveryMail,
536 skel=user_skel,
537 dests=[user_name],
538 recovery_key=recovery_key,
539 user_agent={
540 "device": user_agent.get_device(),
541 "os": user_agent.get_os(),
542 "browser": user_agent.get_browser()
543 }
544 )
546 @exposed
547 @skey(forward_payload="data", session_bound=False)
548 def verify(self, data):
549 def transact(key):
550 skel = self._user_module.editSkel()
551 if not key or not skel.read(key):
552 return None
554 skel["status"] = Status.WAITING_FOR_ADMIN_VERIFICATION \
555 if self.registrationAdminVerificationRequired else Status.ACTIVE
557 skel.write(update_relations=False)
558 return skel
560 if not isinstance(data, dict) or not (skel := db.run_in_transaction(transact, data.get("user_key"))):
561 return self._user_module.render.view(None, tpl=self.verifyFailedTemplate)
563 return self._user_module.render.view(skel, tpl=self.verifySuccessTemplate)
565 def canAdd(self) -> bool:
566 return self.registrationEnabled
568 def addSkel(self) -> skeleton.SkeletonInstance["UserSkel"]:
569 """
570 Prepare the add-Skel for rendering.
571 Currently only calls self._user_module.addSkel() and sets skel["status"] depending on
572 self.registrationEmailVerificationRequired and self.registrationAdminVerificationRequired
573 :return: viur.core.skeleton.Skeleton
574 """
575 skel = self._user_module.addSkel()
577 if self.registrationEmailVerificationRequired:
578 defaultStatusValue = Status.WAITING_FOR_EMAIL_VERIFICATION
579 elif self.registrationAdminVerificationRequired:
580 defaultStatusValue = Status.WAITING_FOR_ADMIN_VERIFICATION
581 else: # No further verification required
582 defaultStatusValue = Status.ACTIVE
584 skel.status.readOnly = True
585 skel["status"] = defaultStatusValue
587 if "password" in skel:
588 skel.password.required = True # The user will have to set a password
590 return skel
592 @force_ssl
593 @exposed
594 @skey(allow_empty=True)
595 def add(self, *, bounce: bool = False, **kwargs):
596 """
597 Allows guests to register a new account if self.registrationEnabled is set to true
599 .. seealso:: :func:`addSkel`, :func:`onAdded`, :func:`canAdd`, :func:`onAdd`
601 :returns: The rendered, added object of the entry, eventually with error hints.
603 :raises: :exc:`viur.core.errors.Unauthorized`, if the current user does not have the required permissions.
604 :raises: :exc:`viur.core.errors.PreconditionFailed`, if the *skey* could not be verified.
605 """
606 if not self.canAdd():
607 raise errors.Unauthorized()
609 skel = self.addSkel()
611 if (
612 not kwargs # no data supplied
613 or not current.request.get().isPostRequest # bail out if not using POST-method
614 or not skel.fromClient(kwargs) # failure on reading into the bones
615 or bounce # review before adding
616 ):
617 # render the skeleton in the version it could as far as it could be read.
618 return self._user_module.render.add(skel)
620 self._user_module.onAdd(skel)
621 skel.write()
623 if self.registrationEmailVerificationRequired and skel["status"] == Status.WAITING_FOR_EMAIL_VERIFICATION:
624 # The user will have to verify his email-address. Create a skey and send it to his address
625 skey = securitykey.create(duration=datetime.timedelta(days=7), session_bound=False,
626 user_key=db.normalize_key(skel["key"]),
627 name=skel["name"])
628 skel.skey = BaseBone(descr="Skey")
629 skel["skey"] = skey
630 email.send_email(dests=[skel["name"]], tpl=self.verifyEmailAddressMail, skel=skel)
632 self._user_module.onAdded(skel) # Call onAdded on our parent user module
633 return self._user_module.render.addSuccess(skel)
636class GoogleAccount(UserPrimaryAuthentication):
637 METHOD_NAME = "X-VIUR-AUTH-Google-Account"
638 NAME = "Google Account"
640 @classmethod
641 def patch_user_skel(cls, skel_cls):
642 """
643 Modifies the UserSkel to be equipped by a bones required by Google Auth
644 """
645 skel_cls.uid = StringBone(
646 descr="Google UserID",
647 required=False,
648 readOnly=True,
649 unique=UniqueValue(UniqueLockMethod.SameValue, False, "UID already in use"),
650 params={
651 "category": "Authentication",
652 },
653 )
655 skel_cls.sync = BooleanBone(
656 descr="Sync user data with OAuth-based services",
657 defaultValue=True,
658 params={
659 "category": "Authentication",
660 "tooltip":
661 "If set, user data like firstname and lastname is automatically kept"
662 "synchronous with the information stored at the OAuth service provider"
663 "(e.g. Google Login)."
664 },
665 )
667 @exposed
668 @force_ssl
669 @skey(allow_empty=True)
670 def login(self, token: str | None = None, *args, **kwargs):
671 if not conf.user.google_client_id:
672 raise errors.PreconditionFailed("Please configure conf.user.google_client_id!")
674 if not token:
675 request = current.request.get()
676 request.response.headers["Content-Type"] = "text/html"
677 if request.response.headers.get("cross-origin-opener-policy") == "same-origin":
678 # We have to allow popups here
679 request.response.headers["cross-origin-opener-policy"] = "same-origin-allow-popups"
681 file_path = conf.instance.core_base_path.joinpath("viur/core/template/vi_user_google_login.html")
682 with open(file_path) as file:
683 tpl_string = file.read()
685 # FIXME: Use Jinja2 for rendering?
686 tpl_string = tpl_string.replace("{{ clientID }}", conf.user.google_client_id)
687 extendCsp({
688 "script-src": ["sha256-JpzaUIxV/gVOQhKoDLerccwqDDIVsdn1JclA6kRNkLw="],
689 "style-src": ["sha256-FQpGSicYMVC5jxKGS5sIEzrRjSJmkxKPaetUc7eamqc="]
690 })
691 return tpl_string
693 user_info = id_token.verify_oauth2_token(token, requests.Request(), conf.user.google_client_id)
694 if user_info["iss"] not in {"accounts.google.com", "https://accounts.google.com"}:
695 raise ValueError("Invalid issuer")
697 # Token looks valid :)
698 uid = user_info["sub"]
699 email = user_info["email"]
701 base_skel = self._user_module.baseSkel()
702 update = False
703 if not (user_skel := base_skel.all().filter("uid =", uid).getSkel()):
704 # We'll try again - checking if there's already an user with that email
705 if not (user_skel := base_skel.all().filter("name.idx =", email.lower()).getSkel()):
706 # Still no luck - it's a completely new user
707 if not self.registrationEnabled:
708 if (domain := user_info.get("hd")) and domain in conf.user.google_gsuite_domains:
709 logging.debug(f"Google user is from allowed {domain} - adding account")
710 else:
711 logging.debug(f"Google user is from {domain} - denying registration")
712 raise errors.Forbidden("Registration for new users is disabled")
714 user_skel = base_skel
715 user_skel["uid"] = uid
716 user_skel["name"] = email
717 update = True
719 # Take user information from Google, if wanted!
720 if user_skel["sync"]:
721 for target, source in {
722 "name": email,
723 "firstname": user_info.get("given_name"),
724 "lastname": user_info.get("family_name"),
725 }.items():
727 if user_skel[target] != source:
728 user_skel[target] = source
729 update = True
731 if update:
732 assert user_skel.write()
734 return self.next_or_finish(user_skel)
737class UserSecondFactorAuthentication(UserAuthentication, abc.ABC):
738 """Abstract class for all second factors."""
739 MAX_RETRY = 3
740 second_factor_login_template = "user_login_secondfactor"
741 """Template to enter the TOPT on login"""
743 @property
744 @abc.abstractmethod
745 def NAME(self) -> str:
746 """Name for this factor for templates."""
747 ...
749 @property
750 @abc.abstractmethod
751 def ACTION_NAME(self) -> str:
752 """The action name for this factor, used as path-segment."""
753 ...
755 def __init__(self, moduleName, modulePath, _user_module):
756 super().__init__(moduleName, modulePath, _user_module)
757 self.action_url = f"{self.modulePath}/{self.ACTION_NAME}"
758 self.add_url = f"{self.modulePath}/add"
759 self.start_url = f"{self.modulePath}/start"
762class TimeBasedOTP(UserSecondFactorAuthentication):
763 METHOD_NAME = "X-VIUR-2FACTOR-TimeBasedOTP"
764 WINDOW_SIZE = 5
765 ACTION_NAME = "otp"
766 NAME = "Time-based OTP"
767 second_factor_login_template = "user_login_secondfactor"
769 @dataclasses.dataclass
770 class OtpConfig:
771 """
772 This dataclass is used to provide an interface for a OTP token
773 algorithm description that is passed within the TimeBasedOTP
774 class for configuration.
775 """
776 secret: str
777 timedrift: float = 0.0
778 algorithm: t.Literal["sha1", "sha256"] = "sha1"
779 interval: int = 60
781 class OtpSkel(skeleton.RelSkel):
782 """
783 This is the Skeleton used to ask for the OTP token.
784 """
785 otptoken = NumericBone(
786 descr="Token",
787 required=True,
788 max=999999,
789 min=0,
790 )
792 @classmethod
793 def patch_user_skel(cls, skel_cls):
794 """
795 Modifies the UserSkel to be equipped by a bones required by Timebased OTP
796 """
797 # One-Time Password Verification
798 skel_cls.otp_serial = StringBone(
799 descr="OTP serial",
800 searchable=True,
801 params={
802 "category": "Second Factor Authentication",
803 },
804 )
806 skel_cls.otp_secret = CredentialBone(
807 descr="OTP secret",
808 params={
809 "category": "Second Factor Authentication",
810 },
811 )
813 skel_cls.otp_timedrift = NumericBone(
814 descr="OTP time drift",
815 readOnly=True,
816 defaultValue=0,
817 precision=1,
818 params={
819 "category": "Second Factor Authentication",
820 },
821 )
823 def get_config(self, skel: skeleton.SkeletonInstance) -> OtpConfig | None:
824 """
825 Returns an instance of self.OtpConfig with a provided token configuration,
826 or None when there is no appropriate configuration of this second factor handler available.
827 """
829 if otp_secret := skel.dbEntity.get("otp_secret"):
830 return self.OtpConfig(secret=otp_secret, timedrift=skel.dbEntity.get("otp_timedrift") or 0)
832 return None
834 def can_handle(self, skel: skeleton.SkeletonInstance) -> bool:
835 """
836 Specified whether the second factor authentication can be handled by the given user or not.
837 """
838 return bool(self.get_config(skel))
840 @exposed
841 def start(self):
842 """
843 Configures OTP login for the current session.
845 A special otp_user_conf has to be specified as a dict, which is stored into the session.
846 """
847 session = current.session.get()
849 if not (user_key := session.get("possible_user_key")):
850 raise errors.PreconditionFailed(
851 "Second factor can only be triggered after successful primary authentication."
852 )
854 user_skel = self._user_module.baseSkel()
855 if not user_skel.read(user_key):
856 raise errors.NotFound("The previously authenticated user is gone.")
858 if not (otp_user_conf := self.get_config(user_skel)):
859 raise errors.PreconditionFailed("This second factor is not available for the user")
861 otp_user_conf = {
862 "key": str(user_key),
863 } | dataclasses.asdict(otp_user_conf)
865 session = current.session.get()
866 session["_otp_user"] = otp_user_conf
867 session.markChanged()
869 return self._user_module.render.edit(
870 self.OtpSkel(),
871 params={
872 "name": i18n.translate(
873 f"viur.core.modules.user.{self.ACTION_NAME}",
874 default_variables={"name": self.NAME},
875 ),
876 "action_name": self.ACTION_NAME,
877 "action_url": f"{self.modulePath}/{self.ACTION_NAME}",
878 },
879 tpl=self.second_factor_login_template
880 )
882 @exposed
883 @force_ssl
884 @skey(allow_empty=True)
885 def otp(self, *args, **kwargs):
886 """
887 Performs the second factor validation and interaction with the client.
888 """
889 session = current.session.get()
890 if not (otp_user_conf := session.get("_otp_user")):
891 raise errors.PreconditionFailed("No OTP process started in this session")
893 # Check if maximum second factor verification attempts
894 if (attempts := otp_user_conf.get("attempts") or 0) > self.MAX_RETRY:
895 raise errors.Forbidden("Maximum amount of authentication retries exceeded")
897 # Read the OTP token via the skeleton, to obtain a valid value
898 skel = self.OtpSkel()
899 if skel.fromClient(kwargs):
900 # Verify the otptoken. If valid, this returns the current timedrift index for this hardware OTP.
901 res = self.verify(
902 otp=skel["otptoken"],
903 secret=otp_user_conf["secret"],
904 algorithm=otp_user_conf.get("algorithm") or "sha1",
905 interval=otp_user_conf.get("interval") or 60,
906 timedrift=otp_user_conf.get("timedrift") or 0.0,
907 valid_window=self.WINDOW_SIZE
908 )
909 else:
910 res = None
912 # Check if Token is invalid. Caution: 'if not verifyIndex' gets false positive for verifyIndex === 0!
913 if res is None:
914 otp_user_conf["attempts"] = attempts + 1
915 session.markChanged()
916 skel.errors = [ReadFromClientError(ReadFromClientErrorSeverity.Invalid, "Wrong OTP Token", ["otptoken"])]
917 return self._user_module.render.edit(
918 skel,
919 name=i18n.translate(
920 f"viur.core.modules.user.auth.{self.ACTION_NAME}",
921 default_variables={"name": self.NAME},
922 ),
923 action_name=self.ACTION_NAME,
924 action_url=f"{self.modulePath}/{self.ACTION_NAME}",
925 tpl=self.second_factor_login_template
926 )
928 # Remove otp user config from session
929 user_key = db.key_helper(otp_user_conf["key"], self._user_module._resolveSkelCls().kindName)
930 del session["_otp_user"]
931 session.markChanged()
933 # Check if the OTP device has a time drift
935 timedriftchange = float(res) - otp_user_conf["timedrift"]
936 if abs(timedriftchange) > 2:
937 # The time-drift change accumulates to more than 2 minutes (for interval==60):
938 # update clock-drift value accordingly
939 self.updateTimeDrift(user_key, timedriftchange)
941 # Continue with authentication
942 return self._user_module.secondFactorSucceeded(self, user_key)
944 @staticmethod
945 def verify(
946 otp: str | int,
947 secret: str,
948 algorithm: str = "sha1",
949 interval: int = 60,
950 timedrift: float = 0.0,
951 for_time: datetime.datetime | None = None,
952 valid_window: int = 0,
953 ) -> int | None:
954 """
955 Verifies the OTP passed in against the current time OTP.
957 This is a fork of pyotp.verify. Rather than true/false, if valid_window > 0, it returns the index for which
958 the OTP value obtained by pyotp.at(for_time=time.time(), counter_offset=index) equals the current value shown
959 on the hardware token generator. This can be used to store the time drift of a given token generator.
961 :param otp: the OTP token to check against
962 :param secret: The OTP secret
963 :param algorithm: digest function to use in the HMAC (expected to be sha1 or sha256)
964 :param interval: the time interval in seconds for OTP. This defaults to 60 (old OTP c200 Generators).
965 :param timedrift: The known timedrift (old index) of the hardware OTP generator
966 :param for_time: Time to check OTP at (defaults to now)
967 :param valid_window: extends the validity to this many counter ticks before and after the current one
969 :returns: The index where verification succeeded, None otherwise
970 """
971 # get the hashing digest
972 digest = {
973 "sha1": hashlib.sha1,
974 "sha256": hashlib.sha256,
975 }.get(algorithm)
977 if not digest:
978 raise errors.NotImplemented(f"{algorithm=} is not implemented")
980 if for_time is None:
981 for_time = datetime.datetime.now()
983 # Timedrift is updated only in fractions in order to prevent problems, but we need an integer index
984 timedrift = round(timedrift)
985 secret = bytes.decode(base64.b32encode(bytes.fromhex(secret))) # decode secret
986 otp = str(otp).zfill(6) # fill with zeros in front
988 # logging.debug(f"TimeBasedOTP:verify: {digest=}, {interval=}, {valid_window=}")
989 totp = pyotp.TOTP(secret, digest=digest, interval=interval)
991 if valid_window:
992 for offset in range(timedrift - valid_window, timedrift + valid_window + 1):
993 token = str(totp.at(for_time, offset))
994 # logging.debug(f"TimeBasedOTP:verify: {offset=}, {otp=}, {token=}")
995 if hmac.compare_digest(otp, token):
996 return offset
998 return None
1000 return 0 if hmac.compare_digest(otp, str(totp.at(for_time, timedrift))) else None
1002 # FIXME: VIUR4 rename
1003 def updateTimeDrift(self, user_key: db.Key, idx: float) -> None:
1004 """
1005 Updates the clock-drift value.
1007 The value is only changed in 1/10 steps, so that a late submit by an user doesn't skew
1008 it out of bounds. Maximum change per call is 0.3 minutes.
1010 :param user_key: For which user should the update occour
1011 :param idx: How many steps before/behind was that token
1012 """
1013 if user_skel := self._user_module.skel().read(user_key):
1014 if otp_skel := self._get_otptoken(user_skel):
1015 otp_skel.patch(
1016 {
1017 "+otp_timedrift": min(max(0.1 * idx, -0.3), 0.3)
1018 },
1019 update_relations=False,
1020 )
1023class AuthenticatorOTP(UserSecondFactorAuthentication):
1024 """
1025 This class handles the second factor for apps like authy and so on
1026 """
1027 METHOD_NAME = "X-VIUR-2FACTOR-AuthenticatorOTP"
1029 # second_factor_add_template = "user_secondfactor_add"
1030 # """Template to configure (add) a new TOPT"""
1032 ACTION_NAME = "authenticator_otp"
1033 """Action name provided for *otp_template* on login"""
1035 NAME = "Authenticator App"
1037 # FIXME: The second factor add has to be rewritten entirely to ActionSkel paradigm.
1038 '''
1039 @exposed
1040 @force_ssl
1041 @skey(allow_empty=True)
1042 def add(self, otp=None):
1043 """
1044 We try to read the otp_app_secret from the current session. When this fails we generate a new one and store
1045 it in the session.
1047 If an otp and a skey are provided we are validate the skey and the otp. If both is successfully we store
1048 the otp_app_secret from the session in the user entry.
1049 """
1050 current_session = current.session.get()
1052 if not (otp_app_secret := current_session.get("_maybe_otp_app_secret")):
1053 otp_app_secret = AuthenticatorOTP.generate_otp_app_secret()
1054 current_session["_maybe_otp_app_secret"] = otp_app_secret
1055 current_session.markChanged()
1057 if otp is None:
1058 return self._user_module.render.second_factor_add(
1059 tpl=self.second_factor_add_template,
1060 action_name=self.ACTION_NAME,
1061 name=i18n.translate(
1062 f"viur.core.modules.user.auth{self.ACTION_NAME}",
1063 default_variables={"name": self.NAME},
1064 ),
1065 add_url=self.add_url,
1066 otp_uri=AuthenticatorOTP.generate_otp_app_secret_uri(otp_app_secret))
1067 else:
1068 if not AuthenticatorOTP.verify_otp(otp, otp_app_secret):
1069 return self._user_module.render.second_factor_add(
1070 tpl=self.second_factor_add_template,
1071 action_name=self.ACTION_NAME,
1072 name=i18n.translate(
1073 f"viur.core.modules.user.auth.{self.ACTION_NAME}",
1074 default_variables={"name": self.NAME},
1075 ),
1076 add_url=self.add_url,
1077 otp_uri=AuthenticatorOTP.generate_otp_app_secret_uri(otp_app_secret)) # to add errors
1079 # Now we can set the otp_app_secret to the current User and render der Success-template
1080 AuthenticatorOTP.set_otp_app_secret(otp_app_secret)
1081 return self._user_module.render.second_factor_add_success(
1082 action_name=self.ACTION_NAME,
1083 name=i18n.translate(
1084 f"viur.core.modules.user.auth.{self.ACTION_NAME}",
1085 default_variables={"name": self.NAME},
1086 ),
1087 )
1088 '''
1090 def can_handle(self, skel: skeleton.SkeletonInstance) -> bool:
1091 """
1092 We can only handle the second factor if we have stored an otp_app_secret before.
1093 """
1094 return bool(skel.dbEntity.get("otp_app_secret", ""))
1096 @classmethod
1097 def patch_user_skel(cls, skel_cls):
1098 """
1099 Modifies the UserSkel to be equipped by bones required by Authenticator App
1100 """
1101 # Authenticator OTP Apps (like Authy)
1102 skel_cls.otp_app_secret = CredentialBone(
1103 descr="OTP Secret (App-Key)",
1104 params={
1105 "category": "Second Factor Authentication",
1106 },
1107 )
1109 @classmethod
1110 def set_otp_app_secret(cls, otp_app_secret=None):
1111 """
1112 Write a new OTP Token in the current user entry.
1113 """
1114 if otp_app_secret is None:
1115 logging.error("No 'otp_app_secret' is provided")
1116 raise errors.PreconditionFailed("No 'otp_app_secret' is provided")
1117 if not (cuser := current.user.get()):
1118 raise errors.Unauthorized()
1120 def transaction(user_key):
1121 if not (user := db.get(user_key)):
1122 raise errors.NotFound()
1123 user["otp_app_secret"] = otp_app_secret
1124 db.put(user)
1126 db.run_in_transaction(transaction, cuser["key"])
1128 @classmethod
1129 def generate_otp_app_secret_uri(cls, otp_app_secret) -> str:
1130 """
1131 :return an otp uri like otpauth://totp/Example:alice@google.com?secret=ABCDEFGH1234&issuer=Example
1132 """
1133 if not (cuser := current.user.get()):
1134 raise errors.Unauthorized()
1135 if not (issuer := conf.user.otp_issuer):
1136 logging.warning(
1137 f"conf.user.otp_issuer is None we replace the issuer by {conf.instance.project_id=}")
1138 issuer = conf.instance.project_id
1140 return pyotp.TOTP(otp_app_secret).provisioning_uri(name=cuser["name"], issuer_name=issuer)
1142 @classmethod
1143 def generate_otp_app_secret(cls) -> str:
1144 """
1145 Generate a new OTP Secret
1146 :return an otp
1147 """
1148 return pyotp.random_base32()
1150 @classmethod
1151 def verify_otp(cls, otp: str | int, secret: str) -> bool:
1152 return pyotp.TOTP(secret).verify(otp)
1154 @exposed
1155 def start(self):
1156 otp_user_conf = {"attempts": 0}
1157 session = current.session.get()
1158 session["_otp_user"] = otp_user_conf
1159 session.markChanged()
1160 return self._user_module.render.edit(
1161 TimeBasedOTP.OtpSkel(),
1162 params={
1163 "name": i18n.translate(
1164 f"viur.core.modules.user.auth.{self.ACTION_NAME}",
1165 default_variables={"name": self.NAME},
1166 ),
1167 "action_name": self.ACTION_NAME,
1168 "action_url": self.action_url,
1169 },
1170 tpl=self.second_factor_login_template,
1171 )
1173 @exposed
1174 @force_ssl
1175 @skey
1176 def authenticator_otp(self, **kwargs):
1177 """
1178 We verify the otp here with the secret we stored before.
1179 """
1180 session = current.session.get()
1181 user_key = db.Key(self._user_module.kindName, session["possible_user_key"])
1183 if not (otp_user_conf := session.get("_otp_user")):
1184 raise errors.PreconditionFailed("No OTP process started in this session")
1186 # Check if maximum second factor verification attempts
1187 if (attempts := otp_user_conf.get("attempts") or 0) > self.MAX_RETRY:
1188 raise errors.Forbidden("Maximum amount of authentication retries exceeded")
1190 if not (user := db.get(user_key)):
1191 raise errors.NotFound()
1193 skel = TimeBasedOTP.OtpSkel()
1194 if not skel.fromClient(kwargs):
1195 raise errors.PreconditionFailed()
1196 otp_token = str(skel["otptoken"]).zfill(6)
1198 if AuthenticatorOTP.verify_otp(otp=otp_token, secret=user["otp_app_secret"]):
1199 return self._user_module.secondFactorSucceeded(self, user_key)
1200 otp_user_conf["attempts"] = attempts + 1
1201 session.markChanged()
1202 skel.errors = [ReadFromClientError(ReadFromClientErrorSeverity.Invalid, "Wrong OTP Token", ["otptoken"])]
1203 return self._user_module.render.edit(
1204 skel,
1205 name=i18n.translate(
1206 f"viur.core.modules.user.auth.{self.ACTION_NAME}",
1207 default_variables={"name": self.NAME},
1208 ),
1209 action_name=self.ACTION_NAME,
1210 action_url=self.action_url,
1211 tpl=self.second_factor_login_template,
1212 )
1215class User(List):
1216 """
1217 The User module is used to manage and authenticate users in a ViUR system.
1219 It is used in almost any ViUR project, but ViUR can also function without any user capabilites.
1220 """
1222 kindName = "user"
1223 addTemplate = "user_add"
1224 addSuccessTemplate = "user_add_success"
1226 authenticationProviders: t.Iterable[UserPrimaryAuthentication] = tuple(filter(
1227 None, (
1228 UserPassword,
1229 conf.user.google_client_id and GoogleAccount,
1230 )
1231 ))
1232 """
1233 Specifies primary authentication providers that are made available
1234 as sub-modules under `user/auth_<classname>`. They might require
1235 customization or configuration.
1236 """
1238 secondFactorProviders: t.Iterable[UserSecondFactorAuthentication] = (
1239 TimeBasedOTP,
1240 AuthenticatorOTP,
1241 )
1242 """
1243 Specifies secondary authentication providers that are made available
1244 as sub-modules under `user/f2_<classname>`. They might require
1245 customization or configuration, which is determined during the
1246 login-process depending on the user that wants to login.
1247 """
1249 validAuthenticationMethods = tuple(filter(
1250 None, (
1251 (UserPassword, AuthenticatorOTP),
1252 (UserPassword, TimeBasedOTP),
1253 (UserPassword, None),
1254 (GoogleAccount, None) if conf.user.google_client_id else None,
1255 )
1256 ))
1257 """
1258 Specifies the possible combinations of primary- and secondary factor
1259 login methos.
1261 GoogleLogin defaults to no second factor, as the Google Account can be
1262 secured by a secondary factor. AuthenticatorOTP and TimeBasedOTP are only
1263 handled when there is a user-dependent configuration available.
1264 """
1266 msg_missing_second_factor = "Second factor required but not configured for this user."
1268 secondFactorTimeWindow = datetime.timedelta(minutes=10)
1270 default_order = "name.idx"
1272 roles = {
1273 "admin": "*",
1274 }
1276 def __init__(self, moduleName, modulePath):
1277 for provider in self.authenticationProviders:
1278 assert issubclass(provider, UserPrimaryAuthentication)
1279 name = f"auth_{provider.__name__.lower()}"
1280 setattr(self, name, provider(name, f"{modulePath}/{name}", self))
1282 for provider in self.secondFactorProviders:
1283 assert issubclass(provider, UserSecondFactorAuthentication)
1284 name = f"f2_{provider.__name__.lower()}"
1285 setattr(self, name, provider(name, f"{modulePath}/{name}", self))
1287 super().__init__(moduleName, modulePath)
1289 def adminInfo(self):
1290 ret = {
1291 "icon": "person-fill",
1292 }
1294 if self.is_admin(current.user.get()):
1295 ret |= {
1296 "actions": [
1297 "trigger_kick",
1298 "trigger_takeover",
1299 ],
1300 "customActions": {
1301 "trigger_kick": {
1302 "name": i18n.translate(
1303 key="viur.core.modules.user.customActions.kick",
1304 defaultText="Kick user",
1305 hint="Title of the kick user function"
1306 ),
1307 "icon": "trash2-fill",
1308 "action": "fetch",
1309 "url": "/vi/{{module}}/trigger/kick/{{key}}",
1310 "confirm": i18n.translate(
1311 key="viur.core.modules.user.customActions.kick.confirm",
1312 defaultText="Do you really want to drop all sessions of the selected user from the system?",
1313 ),
1314 "success": i18n.translate(
1315 key="viur.core.modules.user.customActions.kick.success",
1316 defaultText="Sessions of the user are being invalidated.",
1317 ),
1318 },
1319 "trigger_takeover": {
1320 "name": i18n.translate(
1321 key="viur.core.modules.user.customActions.takeover",
1322 defaultText="Take-over user",
1323 hint="Title of the take user over function"
1324 ),
1325 "icon": "file-person-fill",
1326 "action": "fetch",
1327 "url": "/vi/{{module}}/trigger/takeover/{{key}}",
1328 "confirm": i18n.translate(
1329 key="viur.core.modules.user.customActions.takeover.confirm",
1330 defaultText="Do you really want to replace your current user session by a "
1331 "user session of the selected user?",
1332 ),
1333 "success": i18n.translate(
1334 key="viur.core.modules.user.customActions.takeover.success",
1335 defaultText="You're now know as the selected user!",
1336 ),
1337 "then": "reload-vi",
1338 },
1339 },
1340 }
1342 return ret
1344 def get_role_defaults(self, role: str) -> set[str]:
1345 """
1346 Returns a set of default access rights for a given role.
1348 Defaults to "admin" usage for any role > "user"
1349 and "scriptor" usage for "admin" role.
1350 """
1351 ret = set()
1353 if role in ("viewer", "editor", "admin"):
1354 ret.add("admin")
1356 if role == "admin":
1357 ret.add("scriptor")
1359 return ret
1361 def addSkel(self) -> skeleton.SkeletonInstance["UserSkel"]:
1362 skel = super().addSkel().clone()
1364 if self.is_admin(current.user.get()):
1365 # An admin tries to add a new user.
1366 skel.status.readOnly = False
1367 skel.status.visible = True
1368 skel.access.readOnly = False
1369 skel.access.visible = True
1371 else:
1372 skel.status.readOnly = True
1373 skel["status"] = Status.UNSET
1374 skel.status.visible = False
1375 skel.access.readOnly = True
1376 skel["access"] = []
1377 skel.access.visible = False
1379 if "password" in skel:
1380 # Unlock and require a password
1381 skel.password.required = True
1382 skel.password.visible = True
1383 skel.password.readOnly = False
1385 skel.name.readOnly = False # Don't enforce readonly name in user/add
1386 return skel
1388 def editSkel(self, *args, **kwargs) -> skeleton.SkeletonInstance["UserSkel"]:
1389 skel = super().editSkel().clone()
1391 if "password" in skel:
1392 skel.password.required = False
1393 skel.password.visible = True
1394 skel.password.readOnly = False
1396 lock = not self.is_admin(current.user.get())
1397 skel.name.readOnly = lock
1398 skel.access.readOnly = lock
1399 skel.status.readOnly = lock
1401 return skel
1403 def secondFactorProviderByClass(self, cls) -> UserSecondFactorAuthentication:
1404 return getattr(self, f"f2_{cls.__name__.lower()}")
1406 def getCurrentUser(self):
1407 session = current.session.get()
1409 req = current.request.get()
1410 if session and (session.loaded or req.is_deferred) and (user := session.get("user")):
1411 skel = self.baseSkel()
1412 skel.setEntity(user)
1413 return skel
1415 return None
1417 def continueAuthenticationFlow(self, provider: UserPrimaryAuthentication, user_key: db.Key):
1418 """
1419 Continue authentication flow when primary authentication succeeded.
1420 """
1421 skel = self.baseSkel()
1423 if not skel.read(user_key):
1424 raise errors.NotFound("User was not found.")
1426 if not provider.can_handle(skel):
1427 raise errors.Forbidden("User is not allowed to use this primary login method.")
1429 session = current.session.get()
1430 session["possible_user_key"] = user_key.id_or_name
1431 session["_secondFactorStart"] = utils.utcNow()
1432 session.markChanged()
1434 second_factor_providers = []
1436 for auth_provider, second_factor in self.validAuthenticationMethods:
1437 if isinstance(provider, auth_provider):
1438 if second_factor is not None:
1439 second_factor_provider_instance = self.secondFactorProviderByClass(second_factor)
1440 if second_factor_provider_instance.can_handle(skel):
1441 second_factor_providers.append(second_factor_provider_instance)
1442 else:
1443 second_factor_providers.append(None)
1445 if len(second_factor_providers) > 1 and None in second_factor_providers:
1446 # We have a second factor. So we can get rid of the None
1447 second_factor_providers.pop(second_factor_providers.index(None))
1449 if len(second_factor_providers) == 0:
1450 raise errors.NotAcceptable(self.msg_missing_second_factor)
1451 elif len(second_factor_providers) == 1:
1452 if second_factor_providers[0] is None:
1453 # We allow sign-in without a second factor
1454 return self.authenticateUser(user_key)
1455 # We have only one second factor we don't need the choice template
1456 return second_factor_providers[0].start(user_key)
1458 # In case there is more than one second factor provider remaining, let the user decide!
1459 current.session.get()["_secondfactor_providers"] = {
1460 second_factor.start_url: second_factor.NAME
1461 for second_factor in second_factor_providers
1462 if second_factor.VISIBLE
1463 }
1465 return self.select_secondfactor_provider()
1467 def secondFactorSucceeded(self, provider: UserSecondFactorAuthentication, user_key: db.Key):
1468 """
1469 Continue authentication flow when secondary authentication succeeded.
1470 """
1471 session = current.session.get()
1472 if session["possible_user_key"] != user_key.id_or_name:
1473 raise errors.Forbidden()
1475 # Assert that the second factor verification finished in time
1476 if utils.utcNow() - session["_secondFactorStart"] > self.secondFactorTimeWindow:
1477 raise errors.RequestTimeout()
1479 return self.authenticateUser(user_key)
1481 def is_active(self, skel: skeleton.SkeletonInstance) -> bool | None:
1482 """
1483 Hookable check if a user is defined as "active" and can login.
1485 :param skel: The UserSkel of the user who wants to login.
1486 :returns: Returns True or False when the result is unambigous and the user is active or not. \
1487 Returns None when the provided skel doesn't provide enough information for determination.
1488 """
1489 if skel and "status" in skel:
1490 status = skel["status"]
1491 if not isinstance(status, (Status, int)):
1492 try:
1493 status = int(status)
1494 except ValueError:
1495 status = Status.UNSET
1497 return status >= Status.ACTIVE
1499 return None
1501 def is_admin(self, skel: skeleton.SkeletonInstance) -> bool | None:
1502 """
1503 Hookable check if a user is defined as "admin" and can edit or log into other users.
1504 Defaults to "root" users only.
1506 :param skel: The UserSkel of the user who wants should be checked for user admin privileges.
1507 :returns: Returns True or False when the result is unambigous and the user is admin or not. \
1508 Returns None when the provided skel doesn't provide enough information for determination.
1509 """
1510 if skel and "access" in skel:
1511 return "root" in skel["access"]
1513 return None
1515 def authenticateUser(self, key: db.Key, **kwargs):
1516 """
1517 Performs Log-In for the current session and the given user key.
1519 This resets the current session: All fields not explicitly marked as persistent
1520 by conf.user.session_persistent_fields_on_login are gone afterwards.
1522 :param key: The (DB-)Key of the user we shall authenticate
1523 """
1524 skel = self.baseSkel()
1525 if not skel.read(key):
1526 raise ValueError(f"Unable to authenticate unknown user {key}")
1528 # Verify that this user account is active
1529 if not self.is_active(skel):
1530 raise errors.Forbidden("The user is disabled and cannot be authenticated.")
1532 # Update session for user
1533 session = current.session.get()
1534 # Remember persistent fields...
1535 take_over = {k: v for k, v in session.items() if k in conf.user.session_persistent_fields_on_login}
1536 session.reset()
1537 # and copy them over to the new session
1538 session |= take_over
1540 # Update session, user and request
1541 session["user"] = skel.dbEntity
1543 current.request.get().response.headers[securitykey.SECURITYKEY_STATIC_HEADER] = session.static_security_key
1544 current.user.set(self.getCurrentUser())
1546 self.onLogin(skel)
1548 return self.render.render("login_success", skel, **kwargs)
1551 # Action for primary authentication selection
1553 def SelectAuthenticationProviderSkel(self):
1554 providers = {}
1555 first = None
1556 for provider in self.authenticationProviders:
1557 if not provider.VISIBLE:
1558 continue
1560 provider = getattr(self, f"auth_{provider.__name__.lower()}")
1561 providers[provider.start_url] = provider.NAME
1563 if first is None:
1564 first = provider.start_url
1566 class SelectAuthenticationProviderSkel(skeleton.RelSkel):
1567 provider = SelectBone(
1568 descr="Authentication method",
1569 required=True,
1570 values=providers,
1571 defaultValue=first,
1572 )
1574 return SelectAuthenticationProviderSkel()
1576 @exposed
1577 def select_authentication_provider(self, **kwargs):
1578 skel = self.SelectAuthenticationProviderSkel()
1580 # Read required bones from client
1581 if len(skel.provider.values) > 1 and (not kwargs or not skel.fromClient(kwargs)):
1582 return self.render.render("select_authentication_provider", skel)
1584 return self.render.render("select_authentication_provider_success", skel, next_url=skel["provider"])
1586 # Action for second factor select
1588 class SelectSecondFactorProviderSkel(skeleton.RelSkel):
1589 provider = SelectBone(
1590 descr="Second factor",
1591 required=True,
1592 values=lambda: current.session.get()["_secondfactor_providers"] or (),
1593 )
1595 @exposed
1596 def select_secondfactor_provider(self, **kwargs):
1597 skel = self.SelectSecondFactorProviderSkel()
1599 # Read required bones from client
1600 if not kwargs or not skel.fromClient(kwargs):
1601 return self.render.render("select_secondfactor_provider", skel)
1603 del current.session.get()["_secondfactor_providers"]
1605 return self.render.render("select_secondfactor_provider_success", skel, next_url=skel["provider"])
1607 @exposed
1608 @skey
1609 def logout(self, **kwargs):
1610 """
1611 Implements the logout action. It also terminates the current session (all keys not listed
1612 in viur.session_persistent_fields_on_logout will be lost).
1613 """
1614 if not (user := current.user.get()):
1615 raise errors.Unauthorized()
1617 self.onLogout(user)
1619 session = current.session.get()
1621 if take_over := {k: v for k, v in session.items() if k in conf.user.session_persistent_fields_on_logout}:
1622 session.reset()
1623 session |= take_over
1624 else:
1625 session.clear()
1627 current.user.set(None) # set user to none in context var
1629 return self.render.render("logout_success")
1631 @exposed
1632 def login(self, *args, **kwargs):
1633 return self.select_authentication_provider()
1635 def onLogin(self, skel: skeleton.SkeletonInstance):
1636 """
1637 Hook to be called on user login.
1638 """
1639 # Update the lastlogin timestamp (if available!)
1640 if "lastlogin" in skel:
1641 now = utils.utcNow()
1643 # Conserve DB-Writes: Update the user max once in 30 Minutes (why??)
1644 if not skel["lastlogin"] or ((now - skel["lastlogin"]) > datetime.timedelta(minutes=30)):
1645 skel["lastlogin"] = now
1646 skel.write(update_relations=False)
1648 logging.info(f"""User {skel["name"]} logged in""")
1650 def onLogout(self, skel: skeleton.SkeletonInstance):
1651 """
1652 Hook to be called on user logout.
1653 """
1654 logging.info(f"""User {skel["name"]} logged out""")
1656 @exposed
1657 def view(self, key: db.KeyType = "self", *args, **kwargs):
1658 """
1659 Allow a special key "self" to reference the current user.
1661 By default, any authenticated user can view its own user entry,
1662 to obtain access rights and any specific user information.
1663 This behavior is defined in the customized `canView` function,
1664 which is overwritten by the User-module.
1666 The rendered skeleton can be modified or restriced by specifying
1667 a customized view-skeleton.
1668 """
1669 if key == "self":
1670 if user := current.user.get():
1671 key = user["key"]
1672 else:
1673 raise errors.Unauthorized("Cannot view 'self' with unknown user")
1675 return super().view(key, *args, **kwargs)
1677 def canView(self, skel) -> bool:
1678 if user := current.user.get():
1679 if skel["key"] == user["key"]:
1680 return True
1682 if self.is_admin(user) or "user-view" in user["access"]:
1683 return True
1685 return False
1687 @exposed
1688 @skey(allow_empty=True)
1689 def edit(self, key: db.KeyType = "self", *args, **kwargs):
1690 """
1691 Allow a special key "self" to reference the current user.
1693 This modification will only allow to use "self" as a key;
1694 The specific access right to let the user edit itself must
1695 still be customized.
1697 The rendered and editable skeleton can be modified or restriced
1698 by specifying a customized edit-skeleton.
1699 """
1700 if key == "self":
1701 if user := current.user.get():
1702 key = user["key"]
1703 else:
1704 raise errors.Unauthorized("Cannot edit 'self' with unknown user")
1706 return super().edit(key, *args, **kwargs)
1708 @exposed
1709 def getAuthMethods(self, *args, **kwargs):
1710 """Legacy method prior < viur-core 3.8: Inform tools like Admin which authentication to use"""
1711 logging.warning("DEPRECATED!!! Use '/user/login'-method for this, or update your admin version!")
1713 res = [
1714 (primary.METHOD_NAME, secondary.METHOD_NAME if secondary else None)
1715 for primary, secondary in self.validAuthenticationMethods
1716 ]
1718 return json.dumps(res)
1720 @exposed
1721 def trigger(self, action: str, key: str):
1722 # Check for provided access right definition (equivalent to client-side check), fallback to root!
1723 access = self.adminInfo().get("customActions", {}).get(f"trigger_{action}", {}).get("access") or ()
1724 if not (
1725 (cuser := current.user.get())
1726 and (
1727 any(role in cuser["access"] for role in access)
1728 or self.is_admin(cuser)
1729 )
1730 ):
1731 raise errors.Unauthorized()
1733 skel = self.skel()
1734 if not skel.read(key) and not (skel := skel.all().mergeExternalFilter({"name": key}).getSkel()):
1735 raise errors.NotFound("The provided user does not exist.")
1737 match action:
1738 case "takeover":
1739 self.authenticateUser(skel["key"])
1741 case "kick":
1742 session.killSessionByUser(skel["key"])
1744 case _:
1745 raise errors.NotImplemented(f"Action {action!r} not implemented")
1747 return self.render.render(f"trigger/{action}Success", skel)
1749 @exposed
1750 @access("admin", "root", offer_login=True)
1751 def get_cookie_for_app(self, redirect_to: str = None):
1752 """
1753 Generates a session cookie for the currently logged-in user and hands it to an external
1754 client (script, native app, or WebView).
1756 This endpoint is the entry point of a *App Login Flow*. A privileged user
1757 (admin/root) authenticates normally in the browser and then opens this URL.
1758 The backend creates a fresh ViUR session (see :meth:`_get_cookie_for_app`) and
1759 delivers the resulting ``Set-Cookie`` string to the caller.
1761 **Typical usage — local Python client / script:**
1763 The caller spins up a temporary local HTTP server (e.g. on ``http://localhost:60000``)
1764 and passes its address as *redirect_to*::
1766 /vi/user/get_cookie_for_app?redirect_to=http://localhost:60000
1768 After the user authenticates in the browser, the backend redirects to::
1770 http://localhost:60000?cookie=<url-encoded Set-Cookie string>&app=<project_id>
1772 The local server can then extract only the ``name=value`` part of the cookie string
1773 (everything before the first ``;``) and use it for subsequent API calls::
1775 cookie_str = qs["cookie"][0] # full Set-Cookie value
1776 key, value = cookie_str.split(";", 1)[0].split("=")
1777 session.cookies.update({key: value})
1779 The ``app`` query parameter is set by the server to ``conf.instance.project_id`` and
1780 lets the client distinguish between multiple backends / cache credentials per project.
1782 **Alternative — WebView / browser redirect:**
1784 When the receiving side is a browser or WebView the full ``Set-Cookie`` string can be
1785 forwarded to :meth:`apply_login_cookie` to let the framework activate the session.
1787 :param redirect_to: Optional callback URL. When provided the caller is redirected to
1788 that URL with two query parameters appended automatically:
1790 - ``cookie`` – URL-encoded ``Set-Cookie`` string (``name=value;flags…``).
1791 - ``app`` – the server's GCP project ID (``conf.instance.project_id``).
1793 A ``?`` is appended if the URL does not already contain one.
1794 When omitted, the raw ``Set-Cookie`` string is returned as ``text/plain``
1795 (useful for debugging or direct API calls).
1797 .. warning:: **Open-redirect / session-hijacking risk**
1799 Because the session cookie is appended as a plain query parameter, an
1800 attacker who can convince an authenticated admin to click a crafted link
1801 (e.g. via phishing) could redirect the browser to an evil server that
1802 simply harvests the ``cookie`` parameter and gains full session access.
1804 To mitigate this, all ``redirect_to`` values are validated against
1805 :attr:`conf.user.redirect_whitelist` using :func:`fnmatch.fnmatch`.
1806 Only explicitly whitelisted URL patterns are accepted;
1807 anything else is rejected with ``403 Forbidden``.
1808 Configure the whitelist in your project to include every legitimate
1809 callback origin (local scripts, internal tooling, etc.).
1811 :raises errors.Forbidden: When *redirect_to* does not match any pattern in
1812 :attr:`conf.user.redirect_whitelist`.
1813 :raises errors.Redirect: Always raised when *redirect_to* is supplied and allowed.
1814 """
1815 if redirect_to:
1816 whitelist = utils.ensure_iterable(conf.user.redirect_whitelist)
1817 if not any(fnmatch.fnmatch(redirect_to, pat) for pat in whitelist):
1818 raise errors.Forbidden(f"Redirect target is not whitelisted")
1819 if "?" not in redirect_to:
1820 redirect_to = f"{redirect_to}?"
1821 raise errors.Redirect(
1822 f"{redirect_to}"
1823 f"&cookie={urllib.parse.quote_plus(self._get_cookie_for_app())}"
1824 f"&app={conf.instance.project_id}"
1825 )
1826 current.request.get().response.headers["Content-Type"] = "text/plain"
1827 return self._get_cookie_for_app()
1829 def _get_cookie_for_app(self) -> str:
1830 """
1831 Creates a new, standalone ViUR session for the current user and returns the
1832 corresponding ``Set-Cookie`` header value.
1834 Unlike the regular session created during a normal login, this session is intentionally
1835 **not** attached to the current HTTP request. Instead it is persisted directly in
1836 Datastore so that a different HTTP client can pick it up via :meth:`apply_login_cookie`.
1838 The created session entity mirrors the structure of a regular :class:`Session` entry:
1840 - ``data["user"]`` – the full user ``dbEntity`` (needed by the session loader).
1841 - ``data["is_app_session"]``– flag to distinguish app sessions from regular browser sessions.
1842 - ``static_security_key`` – random value, same role as in normal sessions.
1843 - ``lastseen`` – current timestamp so the session is not immediately garbage-
1844 collected.
1845 - ``user`` – stringified user key for server-side user-based queries.
1847 :returns: A ``Set-Cookie`` header value in the form
1848 ``<cookie_name>=<key>;<flags>`` where ``<flags>`` is produced by
1849 :meth:`Session.build_flags` (``Path=/; HttpOnly; SameSite=…; Secure; Max-Age=…``).
1850 """
1851 cookie_key = utils.string.random(42)
1852 db_session = db.Entity(db.Key(Session.kindName, cookie_key))
1853 data = db.Entity()
1854 data["user"] = current.user.get().dbEntity
1855 data["is_app_session"] = True
1856 db_session["data"] = db.fix_unindexable_properties(data)
1857 db_session["static_security_key"] = utils.string.random(42)
1858 db_session["lastseen"] = time.time()
1859 db_session["user"] = str(current.user.get()["key"])
1860 db_session.exclude_from_indexes = {"data"}
1861 db.put(db_session)
1863 # Provide Set-Cookie header entry with configured properties
1864 return f"{Session.cookie_name}={cookie_key};{Session.build_flags()}"
1866 @exposed
1867 def apply_login_cookie(self, cookie: str):
1868 """
1869 Redirect endpoint to load session from the given cookie.
1871 This is the second half of the *App Login Flow*. A native app or WebView that received a
1872 ``Set-Cookie`` string from :meth:`get_cookie_for_app` (typically via a redirect URL
1873 parameter) calls this endpoint to activate the embedded session for its own HTTP context.
1875 The flow is:
1877 1. Parse the raw ``Set-Cookie`` string with :class:`http.cookies.SimpleCookie`.
1878 2. Look for the expected session cookie name (:attr:`Session.cookie_name`).
1879 3. Reset the caller's current (anonymous) session.
1880 4. Inject the cookie value into the current request's cookie jar so that
1881 :meth:`Session.load` can find the pre-built Datastore session.
1882 5. Redirect to ``/`` – from this point on the caller is fully authenticated.
1884 :param cookie: A raw ``Set-Cookie`` header value as produced by :meth:`_get_cookie_for_app`,
1885 e.g. ``viur_cookie_myproject=<key>;Path=/;HttpOnly;…``.
1886 :raises errors.Redirect: On success – redirects to ``/``.
1887 :raises errors.BadRequest: When the cookie string does not contain a recognisable session
1888 cookie (i.e. :attr:`Session.cookie_name` is absent after parsing).
1889 """
1890 cookies = SimpleCookie()
1891 cookies.load(cookie)
1892 if Session.cookie_name in cookies:
1893 session_cookie = cookies[Session.cookie_name]
1894 current.session.get().reset()
1895 current.request.get().request.cookies[session_cookie.key] = session_cookie.value
1896 current.session.get().load()
1897 raise errors.Redirect("/")
1898 else:
1899 raise errors.BadRequest
1901 def onEdited(self, skel):
1902 super().onEdited(skel)
1904 # In case the user is set to inactive, kill all sessions
1905 if self.is_active(skel) is False:
1906 session.killSessionByUser(skel["key"])
1908 # Otherwise, update the user entity cached in all the user's sessions
1909 else:
1910 session.update_session_user(skel["key"])
1912 def onDeleted(self, skel):
1913 super().onDeleted(skel)
1914 # Invalidate all sessions of that user
1915 session.killSessionByUser(skel["key"])
1918@tasks.StartupTask
1919def createNewUserIfNotExists():
1920 """
1921 Create a new Admin user, if the userDB is empty
1922 """
1923 if (
1924 (user_module := getattr(conf.main_app.vi, "user", None))
1925 and isinstance(user_module, User)
1926 and "addSkel" in dir(user_module)
1927 and "validAuthenticationMethods" in dir(user_module)
1928 # UserPassword must be one of the primary login methods
1929 and any(
1930 issubclass(provider[0], UserPassword)
1931 for provider in user_module.validAuthenticationMethods
1932 )
1933 ):
1934 if not db.Query(user_module.addSkel().kindName).getEntry(): # There's currently no user in the database
1935 addSkel = skeleton.skeletonByKind(user_module.addSkel().kindName)() # Ensure we have the full skeleton
1936 uname = f"""admin@{conf.instance.project_id}.appspot.com"""
1937 pw = utils.string.random(13)
1938 addSkel["name"] = uname
1939 addSkel["status"] = Status.ACTIVE # Ensure it's enabled right away
1940 addSkel["access"] = ["root"]
1941 addSkel["password"] = pw
1943 try:
1944 addSkel.write()
1945 except Exception as e:
1946 logging.critical(f"Something went wrong when trying to add admin user {uname!r} with Password {pw!r}")
1947 logging.exception(e)
1948 return
1950 msg = f"ViUR created a new admin-user for you!\nUsername: {uname}\nPassword: {pw}"
1952 logging.warning(msg)
1953 email.send_email_to_admins("New ViUR password", msg)
1956# DEPRECATED ATTRIBUTES HANDLING
1958def __getattr__(attr):
1959 match attr:
1960 case "userSkel": 1960 ↛ 1961line 1960 didn't jump to line 1961 because the pattern on line 1960 never matched
1961 msg = f"Use of `userSkel` is deprecated; Please use `UserSkel` instead!"
1962 warnings.warn(msg, DeprecationWarning, stacklevel=2)
1963 logging.warning(msg)
1964 return UserSkel
1966 return super(__import__(__name__).__class__).__getattr__(attr)