# privacyIDEA is a fork of LinOTP
#
# 2015-11-03 Cornelius Kölbel <cornelius@privacyidea.org>
# Add memberfunction "exist"
# 2015-06-06 Cornelius Kölbel <cornelius@privacyidea.org>
# Add the possibility to update the user data.
# Nov 27, 2014 Cornelius Kölbel <cornelius@privacyidea.org>
# Migration to flask
# Rewrite of methods
# 100% test code coverage
# May 08, 2014 Cornelius Kölbel
#
# License: AGPLv3
# contact: http://www.privacyidea.org
#
# 2014-10-03 fix getUsername function
# Cornelius Kölbel <cornelius@privcyidea.org>
#
# Copyright (C) 2010 - 2014 LSE Leading Security Experts GmbH
# License: AGPLv3
# contact: http://www.linotp.org
# http://www.lsexperts.de
# linotp@lsexperts.de
#
# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This code is free software; you can redistribute it and/or
# modify it under the terms of the GNU AFFERO GENERAL PUBLIC LICENSE
# License as published by the Free Software Foundation; either
# version 3 of the License, or any later version.
#
# This code is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU AFFERO GENERAL PUBLIC LICENSE for more details.
#
# You should have received a copy of the GNU Affero General Public
# License along with this program. If not, see <http://www.gnu.org/licenses/>.
#
__doc__ = '''These are the library functions for user functions.
It depends on the lib.resolver and lib.realm.
There are and must be no dependencies to the token functions (lib.token)
or to webservices!
This code is tested in tests/test_lib_user.py
'''
import hashlib
import logging
import traceback
from typing import Any
from sqlalchemy import select, delete
from privacyidea.lib.cache.user import (cached_user_id, cached_user_info, cached_username,
invalidate_user)
from privacyidea.lib.error import ParameterError, ResolverError, UserError
from privacyidea.models import CustomUserAttribute, InternalUserAttribute, db
from .config import get_from_config, SYSCONF
from .log import log_with
from .realm import (get_realms, realm_is_defined,
get_default_realm,
get_ordered_resolvers,
get_realm_id,
get_realms_of_resolver)
from .resolver import (get_resolver_object,
get_resolver_type)
from .usercache import (user_cache, cache_username, user_init, delete_user_cache)
from privacyidea.lib.params import get_optional, get_required
log = logging.getLogger(__name__)
[docs]
class User:
"""
The user has the attributes ``login``, ``realm`` and ``resolver``.
Usually a user can be found via "login@realm".
A user object with an empty login and realm should not exist,
whereas a user object could have an empty resolver.
"""
# In some test case the login attribute from a not
# initialized user is requested. This is why we need
# these dummy class attributes.
login = ""
realm = ""
resolver = ""
# NOTE: Directly decorating the class ``User`` breaks ``isinstance`` checks,
# which is why we have to decorate __init__
@log_with(log)
def __init__(self, login="", realm="", resolver="", uid=None):
self.login = login or ""
self.used_login = self.login
self.realm = (realm or "").lower()
self.realm_id = None
if resolver == "**":
resolver = ""
self.resolver = resolver or ""
# We never specified the type of the UID, but internally we expect it to be a string (i.e. TokenOwner table)
# To avoid confusion here if an uid parameter is passed we convert it to a string.
self.uid = str(uid) if uid else None
self.rtype = None
# Hash of already checked passwords and their result. If a user has multiple token, it is not necessary to check
# the same password multiple times. However, we can not differentiate between PIN+OTP and just PIN, so this dict
# will have an entry for PIN+OTP (likely to fail) and then the OTP cut off to just PIN. In case the PIN was
# given in the request, the dict will have only one entry.
self._checked_passwords = {}
if not self.login and not self.resolver and uid is not None:
raise UserError("Can not create a user object from a uid without a resolver!")
# Enrich user object with information from the userstore or from the
# user cache
if login or uid is not None:
self._get_user_from_userstore()
# Just store the resolver type
self.rtype = get_resolver_type(self.resolver)
# Add realm_id to User object
self.realm_id = get_realm_id(self.realm)
@user_cache(user_init)
def _get_user_from_userstore(self):
if not self.resolver:
# set the resolver implicitly!
self._get_resolvers()
# Get Identifiers
if self.resolver:
resolver = get_resolver_object(self.resolver)
if resolver is None:
raise UserError(f"The resolver '{self.resolver!s}' does not exist!")
if self.uid is None:
# Determine the uid
self.uid = cached_user_id(resolver, self.resolver, self.login)
if not self.login:
# Determine the login if it does not exist or
self.used_login = self.login = cached_username(resolver, self.resolver, self.uid)
if resolver.has_multiple_loginnames:
# if the resolver has multiple logins the primary login might be another value!
self.login = cached_username(resolver, self.resolver, self.uid)
[docs]
def is_empty(self) -> bool:
# ignore if only resolver is set! as it makes no sense
if len(self.login or "") + len(self.realm or "") == 0:
return True
else:
return False
def __eq__(self, other):
"""
Compare two User Objects.
:param other: The other User object, to which this very object is
compared.
:type other: User object
:return: True or False
:rtype: bool
"""
if not isinstance(other, type(self)):
log.info(f"Comparing a non-user object: {self!s} != {type(other)!s}.")
return False
if (self.resolver != other.resolver) or (self.realm != other.realm):
log.info("Users are not in the same resolver and realm: "
f"{self!s} != {other!s}.")
return False
if self.uid and other.uid:
log.debug(f"Comparing based on uid: {self.uid!s} vs {other.uid!s}")
return self.uid == other.uid
log.debug(f"Comparing based on login: {self.login!s} vs {other.login!s}")
return self.login == other.login
def __ne__(self, other):
"""
Compare two user objects and return true, if they are not equal
:param other: The other User object
:return: True or False
"""
return not self.__eq__(other)
def __hash__(self):
return hash((type(self), self.login, self.resolver, self.realm))
def __str__(self):
ret = "<empty user>"
if not self.is_empty():
# Realm and resolver should always be ASCII
conf = ''
if self.resolver:
conf = f'.{self.resolver!s}'
ret = f'<{self.login!s}{conf!s}@{self.realm!s}>'
return ret
def __repr__(self):
ret = (f"User(login={self.login!r}, realm={self.realm!r}, resolver={self.resolver!r})")
return ret
def __bool__(self):
return not self.is_empty()
__nonzero__ = __bool__
def _get_resolvers(self, all_resolvers=False) -> list[str]:
"""
This returns the list of the resolvernames of the user.
If no resolver attribute exists at the moment, the user is searched
in the realm and according to this the resolver attribute is set.
It will only return one resolver in the list for backward compatibility
.. note:: If the user does not exist in the realm, then an empty
list is returned!
:param all_resolvers: return all resolvers (of a realm), in which
the user is contained
:return: list of resolvers for self.login
"""
if self.resolver:
return [self.resolver]
resolvers = []
for resolver_name in get_ordered_resolvers(self.realm):
# test, if the user is contained in this resolver
if self._locate_user_in_resolver(resolver_name):
break
if self.resolver:
resolvers = [self.resolver]
return resolvers
def _locate_user_in_resolver(self, resolvername: str) -> bool:
"""
Try to locate the user (by self.login) in the resolver with the given name.
In case of success, this sets `self.resolver` as well as `self.uid`
and returns True. If the resolver does not exist or the user does
not exist in the resolver, False is returned.
:param resolvername: string denoting the resolver name
:return: boolean
"""
resolver = get_resolver_object(resolvername)
if resolver is None: # pragma: no cover
log.info(f"Resolver {resolvername!r} not found!")
return False
else:
uid = cached_user_id(resolver, resolvername, self.login)
if uid not in ["", None]:
log.info(f"user {self.login!r} found in resolver {resolvername!r}")
log.info(f"userid resolved to {uid!r} ")
self.resolver = resolvername
self.uid = uid
# We do not need to search other resolvers!
return True
else:
log.debug(f"user {self.login!r} not found"
f" in resolver {resolvername!r}")
return False
[docs]
def get_user_identifiers(self) -> tuple[str or int, str, str]:
"""
This returns the UserId information from the resolver object and
the resolvertype and the resolvername
(former: getUserId)
(former: getUserResolverId)
:return: The userid, the resolver type and the resolver name
like (1000, "passwdresolver", "resolver1")
:rtype: tuple
"""
if not self.resolver:
raise UserError("The user can not be found in any resolver in "
"this realm!")
return self.uid, self.rtype, self.resolver
[docs]
def exist(self) -> bool:
"""
Check if the user object exists in the user store
:return: True or False
"""
# TODO: really check if user exist (ask user store and maybe re-evaluate realm)
exist = self.uid and self.realm_id
return exist
@property
def info(self) -> dict:
"""
return the detailed information for the user
:return: a dict with all the userinformation
:rtype: dict
"""
return self.get_specific_info()
[docs]
def get_specific_info(self, attributes: list[str] = None) -> dict:
"""
returns the specified attributes for the user or all if attributes is None
:return: a dict with the specified user information
"""
if self.is_empty() or not self.exist():
# An empty user has no info
return {}
(uid, _rtype, _resolver) = self.get_user_identifiers()
if uid is None:
return {}
resolver = get_resolver_object(self.resolver)
available_attributes = resolver.get_available_info_keys()
# For now, only exclude groups if not requested as this one might be expensive to retrieve, but request all
# others to not completely break the LDAP cache
if attributes is not None and "groups" not in attributes and "groups" in available_attributes:
available_attributes.remove("groups")
full_user_info = cached_user_info(resolver, self.resolver, uid, available_attributes)
# only return requested attributes
user_info = {key: value for key, value in full_user_info.items() if attributes is None or key in attributes}
# Now add the custom attributes, this is used e.g. in ADDUSERINRESPONSE
user_info.update(self.attributes)
return user_info
@property
def available_info_keys(self) -> list[str]:
"""
returns the possible keys for user information for this user
:return: a list of possible keys for user information
:rtype: list
"""
if self.is_empty() or not self.exist():
# An empty user has no info
return []
resolver = get_resolver_object(self.resolver)
return resolver.get_available_info_keys()
[docs]
@log_with(log)
def set_attribute(self, attribute_key: str, attribute_value: str, attribute_type: str = None) -> int:
"""
Set a custom attribute for a user
:param attribute_key: The key of the attribute
:param attribute_value: The value of the attribute
:return: The id of the attribute setting
"""
stmt = select(CustomUserAttribute).filter_by(
user_id=self.uid,
resolver=self.resolver,
realm_id=self.realm_id,
Key=attribute_key
)
existing_attribute = db.session.execute(stmt).scalar_one_or_none()
if existing_attribute:
existing_attribute.Value = attribute_value
existing_attribute.Type = attribute_type
attribute_id = existing_attribute.id
else:
new_attribute = CustomUserAttribute(user_id=self.uid, resolver=self.resolver, realm_id=self.realm_id,
Key=attribute_key, Value=attribute_value, Type=attribute_type)
db.session.add(new_attribute)
db.session.flush()
attribute_id = new_attribute.id
db.session.commit()
return attribute_id
@property
def attributes(self) -> dict:
"""
returns the custom attributes of a user
:return: a dictionary of attributes with keys and values
"""
return get_attributes(self.uid, self.resolver, self.realm_id)
[docs]
@log_with(log)
def delete_attribute(self, attribute_key: str = None) -> int:
"""
Delete the given key as custom user attribute.
If no key is given, then all attributes are deleted
:param attribute_key: The key to delete
:return: The number of deleted rows
"""
stmt = delete(CustomUserAttribute).filter_by(user_id=self.uid, resolver=self.resolver,
realm_id=self.realm_id)
if attribute_key:
stmt = stmt.filter_by(Key=attribute_key)
result = db.session.execute(stmt)
db.session.commit()
return result.rowcount
def _require_resolved_for_write(self) -> None:
"""
Writes/deletes on internal attributes key on (uid, resolver, realm_id).
An unresolved user has ``uid=''``, so writing under it would create
rows shared by every unresolved user — i.e. cross-user data leak.
We also require ``realm_id``: it is part of the unique constraint, but
SQL treats NULLs as distinct, so a NULL realm_id would silently bypass
the per-user-key dedup backstop. Requiring it here (matching the
``uid and realm_id`` notion of an existing user, see :meth:`exist`)
guarantees every written row has a concrete identity.
Reads are tolerated and return an empty dict (see ``internal_attributes``).
"""
if not self.uid or not self.realm_id:
raise UserError(f"Cannot modify internal attributes for unresolved user "
f"(login={self.login!r}, realm={self.realm!r}).")
@property
def internal_attributes(self) -> dict:
"""
Returns all internal user attributes for this user as a dict.
Internal attributes are written by privacyIDEA itself (e.g. cached
FIDO2 user IDs, last-used token per client) and are NOT meant to be
used in policy conditions. Use :meth:`attributes` for admin-facing data.
For an unresolved user (empty uid) this returns ``{}`` rather than
raising — callers like ``preferred_client_mode`` run for every auth
response, including serial-only flows that legitimately have no user.
"""
if not self.uid:
return {}
return get_internal_attributes(self.uid, self.resolver, self.realm_id)
[docs]
@log_with(log)
def set_internal_attribute(self, key: str, value: Any) -> int:
"""
Set an internal attribute for this user. ``value`` may be any
JSON-serializable Python object.
The model's ``node`` column is reserved for future node-local state
and is intentionally not exposed here: it is part of neither the
lookup nor the unique constraint yet, so writing it would create
inconsistent rows. When node-awareness is implemented, ``node`` must
be added to the unique constraint and to the SELECT below before it
can be written here.
Not safe against concurrent writes on the same ``key`` for the same
user: two callers can both miss the SELECT and race on the INSERT,
with one hitting the UNIQUE constraint. Acceptable for the current
callers (hint-style values, low contention).
"""
self._require_resolved_for_write()
stmt = select(InternalUserAttribute).filter_by(
user_id=self.uid,
resolver=self.resolver,
realm_id=self.realm_id,
Key=key,
)
existing = db.session.execute(stmt).scalar_one_or_none()
if existing:
existing.Value = value
attribute_id = existing.id
else:
new_attribute = InternalUserAttribute(user_id=self.uid, resolver=self.resolver,
realm_id=self.realm_id, Key=key, Value=value)
db.session.add(new_attribute)
db.session.flush()
attribute_id = new_attribute.id
db.session.commit()
return attribute_id
[docs]
@log_with(log)
def delete_internal_attribute(self, key: str | None = None) -> int:
"""
Delete an internal attribute. If ``key`` is None, all internal
attributes for this user are deleted.
"""
self._require_resolved_for_write()
stmt = delete(InternalUserAttribute).filter_by(user_id=self.uid, resolver=self.resolver,
realm_id=self.realm_id)
if key:
stmt = stmt.filter_by(Key=key)
result = db.session.execute(stmt)
db.session.commit()
return result.rowcount
[docs]
@log_with(log)
def get_user_phone(self, phone_type: str = 'phone', index: int = None) -> str or list[str]:
"""
Returns the phone number or a list of phone numbers of a user.
:param phone_type: The type of the phone, i.e. either mobile or
phone (land line)
:type phone_type: string
:param index: The index of the selected phone number of list of the phones of the user.
If the index is given, this phone number as string is returned.
If the index is omitted, all phone numbers are returned.
:returns: list with phone numbers of this user object
"""
userinfo = self.get_specific_info([phone_type])
if phone_type in userinfo:
phone = userinfo[phone_type]
log.debug(f"got user phone {phone!r} of type {phone_type!r}")
if isinstance(phone, list) and index is not None:
if len(phone) > index:
return phone[index]
else:
log.warning(f"userobject ({self!r}) has not that much "
f"phone numbers ({index!r} of {phone!r}).")
return ""
else:
return phone
else:
log.warning(f"userobject ({self!r}) has no phone of type {phone_type!r}.")
return ""
[docs]
@log_with(log)
def get_user_realms(self) -> list[str]:
"""
Returns a list of the realms, a user belongs to.
Usually this will only be one realm.
But if the user object has no realm but only a resolver,
than all realms, containing this resolver are returned.
This function is used for the policy module
:return: realms of the user
:rtype: list
"""
all_realms = get_realms()
user_realms = []
if self.realm == "" and self.resolver == "":
default_realm = get_default_realm().lower()
user_realms.append(default_realm)
self.realm = default_realm
elif self.realm != "":
user_realms.append(self.realm.lower())
else:
# User has no realm!
# we have got a resolver and will get all realms
# the resolver belongs to.
for key, val in all_realms.items():
log.debug(f"evaluating realm {key!r}: {val!r} ")
for reso in val.get('resolver', []):
resoname = reso.get("name")
if resoname == self.resolver:
user_realms.append(key.lower())
log.debug(f"added realm {key!r} to Realms due to "
f"resolver {self.resolver!r}")
return user_realms
[docs]
@log_with(log, log_entry=False)
def check_password(self, password: str) -> str or None:
"""
The password of the user is checked against the user source
:param password: The clear text password
:return: the username of the authenticated user.
If unsuccessful, returns None
"""
success = None
# The password hash is used to avoid multiple password checks at the
# resolver. It is not persisted in any way and only stored in memory for
# the duration of the request.
password_hash = hashlib.sha3_512(password.encode("utf-8")).hexdigest()
try:
log.info(f"User {self.login} from realm {self.realm} tries to authenticate")
# If the password was already checked, return the known result
if password_hash in self._checked_passwords.keys():
if self._checked_passwords[password_hash]:
success = f"{self.login}@{self.realm}"
log.debug(f"Successfully authenticated user {self} from request cache.")
else:
log.info(f"User {self} failed to authenticate from request cache.")
return success
res = self._get_resolvers()
# Now we know, the resolvers of this user, and we can verify the password
if len(res) == 1:
from .resolvers.HTTPResolver import HTTPResolver
resolver = get_resolver_object(self.resolver)
uid, _rtype, _rname = self.get_user_identifiers()
if isinstance(resolver, HTTPResolver):
valid_credentials = resolver.checkPass(uid, password, self.login)
else:
valid_credentials = resolver.checkPass(uid, password)
if valid_credentials:
success = f"{self.login}@{self.realm}"
log.debug(f"Successfully authenticated user {self}.")
self._checked_passwords[password_hash] = True
else:
log.info(f"User {self} failed to authenticate.")
self._checked_passwords[password_hash] = False
elif not res:
log.error(f"The user {self!r} exists in NO resolver.")
except UserError as e: # pragma: no cover
log.error(f"Error while trying to verify the username: {e}")
except Exception as e: # pragma: no cover
log.error(f"Error checking password within module {e}")
log.debug(f"{traceback.format_exc()}")
return success
[docs]
@log_with(log)
def get_search_fields(self) -> dict:
"""
Return the valid search fields of a user.
The search fields are defined in the UserIdResolver class.
:return: searchFields with name (key) and type (value)
:rtype: dict
"""
search_fields = {}
for reso in self._get_resolvers():
# try to load the UserIdResolver Class
try:
y = get_resolver_object(reso)
sf = y.get_search_fields()
search_fields[reso] = sf
except Exception as e: # pragma: no cover
log.warning(f"module {reso!r}: {e!r}")
return search_fields
# If passwords should not be logged, we hide it from the log entry
[docs]
@log_with(log)
def update_user_info(self, attributes : dict, password: str = None) -> bool:
"""
This updates the given attributes of a user.
The attributes can be "username", "surname", "givenname", "email",
"mobile", "phone", "password"
:param attributes: A dictionary of the attributes to be updated
:param password: The password of the user
:return: True in case of success
"""
if password is not None:
attributes["password"] = password
success = False
try:
log.info(f"User info for user {self.login!r}@{self.realm!r} about to "
"be updated.")
res = self._get_resolvers()
# Now we know, the resolvers of this user and we can update the
# user
if len(res) == 1:
resolver = get_resolver_object(self.resolver)
if not resolver.updateable: # pragma: no cover
log.warning(f"The resolver {resolver!r} is not updateable.")
else:
uid, _rtype, _rname = self.get_user_identifiers()
if resolver.update_user(uid, attributes):
success = True
# Delete entries corresponding to the old username from the user cache
delete_user_cache(username=self.login, resolver=self.resolver)
invalidate_user(self.resolver, login=self.login, user_id=uid)
# If necessary, update the username
if attributes.get("username"):
self.login = attributes.get("username")
log.info(f"Successfully updated user {self!r}.")
else: # pragma: no cover
log.info(f"user {self!r} failed to update.")
elif not res: # pragma: no cover
log.error(f"The user {self!r} exists in NO resolver.")
except UserError as exx: # pragma: no cover
log.error(f"Error while trying to verify the username: {exx!r}")
return success
[docs]
@log_with(log)
def delete(self) -> bool:
"""
This deletes the user in the user store. I.e. the user in the SQL
database or the LDAP gets deleted.
Returns True in case of success
"""
success = False
try:
log.info(f"User {self.login!r}@{self.realm!r} about to be deleted.")
res = self._get_resolvers()
# Now we know, the resolvers of this user and we can delete it
if len(res) == 1:
resolver = get_resolver_object(self.resolver)
if not resolver.updateable: # pragma: no cover
log.warning(f"The resolver {resolver!r} is not updateable.")
else:
uid, _rtype, _rname = self.get_user_identifiers()
if resolver.delete_user(uid):
success = True
log.info(f"Successfully deleted user {self!r}.")
# Delete corresponding entry from the user cache
delete_user_cache(username=self.login, resolver=self.resolver)
invalidate_user(self.resolver, login=self.login, user_id=uid)
else: # pragma: no cover
log.info(f"user {self!r} failed to update.")
elif not res: # pragma: no cover
log.error(f"The user {self!r} exists in NO resolver.")
except UserError as exx: # pragma: no cover
log.error(f"Error while trying to verify the username: {exx!r}")
except Exception as exx: # pragma: no cover
log.error(f"Error checking password within module {exx!r}")
log.debug(f"{traceback.format_exc()!s}")
return success
[docs]
def user_export_dict(self) -> dict:
"""
Returns a dictionary with the user identifiers, which can be used to
assign a token to the same user after import.
:return: A dictionary with the user identifiers
"""
return {
"login": self.login,
"realm": self.realm,
"resolver": self.resolver,
"uid": self.uid,
"custom_attributes": self.attributes
}
[docs]
@log_with(log)
def create_user(resolvername: str, attributes: dict, password: str = None) -> int or str:
"""
This creates a new user in the given resolver. The resolver must be
editable to do so.
The attributes is a dictionary containing the keys "username", "email",
"phone",
"mobile", "surname", "givenname", "password".
We return the UID and not the user object, since the user could be located
in several realms!
:param resolvername: The name of the resolver, in which the user should
be created
:type resolvername: basestring
:param attributes: Attributes of the user
:type attributes: dict
:param password: The password of the user
:return: The uid of the user object
"""
if password is not None:
attributes["password"] = password
resolver = get_resolver_object(resolvername)
uid = resolver.add_user(attributes)
return uid
[docs]
@log_with(log)
def split_user(username: str) -> tuple[str, str]:
"""
Split the username of the form user@realm into the username and the realm
splitting myemail@emailprovider.com@realm is also possible and will
return (myemail@emailprovider.com, realm).
If for a user@domain the "domain" does not exist as realm, the name is
not split, since it might be the user@domain in the default realm
If the Split@Sign configuration is disabled, the username won't be split
and the username and an empty realm will be returned.
We can also split realm\\user to (user, realm)
:param username: the username to split
:type username: string
:return: username and realm
:rtype: tuple
"""
user = username.strip()
realm = ""
split_at_sign = get_from_config(SYSCONF.SPLITATSIGN, return_bool=True)
if split_at_sign:
user_split = user.split('@')
if len(user_split) >= 2:
if realm_is_defined(user_split[-1]):
# split the last only if the last part is really a realm
(user, realm) = user.rsplit('@', 1)
else:
user_split = user.split('\\')
if len(user_split) >= 2:
(realm, user) = user.rsplit('\\', 1)
return user, realm
[docs]
@log_with(log)
def get_user_from_param(param: dict, optional_or_required: bool = True) -> User:
"""
Find the parameter user, realm and resolver and
create a user object from these parameters.
An exception is raised, if a user in a realm is found in more
than one resolver.
:param param: The dictionary of request parameters
:param optional_or_required: ``True`` (default) if the user param is optional,
``False`` if it is required (raises ParameterError when absent).
:return: User as found in the parameters
"""
realm = ""
if optional_or_required:
username = get_optional(param, "user")
else:
username = get_required(param, "user")
if username is None:
username = ""
else:
username, realm = split_user(username)
if "realm" in param:
realm = param["realm"]
if username != "":
if realm is None or realm == "":
realm = get_default_realm()
user_object = User(login=username, realm=realm,
resolver=param.get("resolver"))
return user_object
[docs]
@log_with(log)
def get_user_list(param: dict | None = None, user: User | None = None,
include_custom_attributes: bool = False,
requested_attributes: list[str] | None = None,
failures: list[str] | None = None) -> list[dict]:
"""
This function returns a list of user dictionaries. The user dict contains the resolver and custom user attributes,
if requested.
The ``realm`` parameter may be a single realm name, a comma-separated string of realm names, or a list of
realm names (surrounding whitespace is ignored). Scoping by ``realm`` / ``resolver`` in ``param``:
* ``realm`` only: query every resolver in each given realm.
* ``resolver`` only: query only that resolver, in every realm that
contains it.
* ``realm`` and ``resolver``: query only that resolver, and only
within each given realm. If the resolver is not part of a realm, that
realm contributes no results.
* neither: query every resolver in every realm.
An empty ``realm`` or ``resolver`` value is treated as if it were not
supplied and therefore does not restrict the scope. When a ``user`` object
is passed as well, its resolver narrows only the user's own realm, not a
separately requested ``realm``.
The ``realm``, ``resolver`` and ``editable`` keys are added on the lib layer and are only included in the
returned user dictionaries when ``requested_attributes`` is None/empty or explicitly lists them.
:param param: search parameters
:param user: a specific user object to return
:param include_custom_attributes: Set to True, if you want to receive custom attributes of external users.
:param requested_attributes: A list of attributes to return for each user. If None or empty, all
attributes are returned.
:param failures: optional list. When provided, the name of every resolver that could not be
queried is appended once (deduplicated). This covers resolvers that raised ``ResolverError``
or ``ParameterError`` as well as resolvers that are assigned to a realm in scope but pinned to
a different node and could therefore not be queried here. Callers that want to surface partial
failures can pass an empty list and inspect it after the call.
:return: list of user info as dictionaries
"""
# The user dictionary, what we use to avoid duplicates in realms, while searching for users. The key will be the
# tuple of (username, realm)
users_dict = {}
resolvers = []
search_dict = {"username": "*"}
param = param or {}
# we have to recreate a new searchdict without the realm key
# as delete does not work
for key in param:
lval = param[key]
if key in ["realm", "resolver", "user", "username"]:
continue
search_dict[key] = lval
log.debug(f"Parameter key:{key!r}={lval!r}")
# update search_dict depending on existence of 'user' or 'username' in param
# Since 'user' takes precedence over 'username' we have to check the order
if 'username' in param:
search_dict['username'] = param['username']
if 'user' in param:
search_dict['username'] = param['user']
log.debug('Changed search key to username: %s.', search_dict['username'])
# determine which scope we want to show
param_resolver = get_optional(param, "resolver")
param_realm_raw = get_optional(param, "realm")
# param_realm_raw may be a single string, a comma-separated string of
# multiple realms, or a list. Normalise to a list of individual realm
# names (or an empty list when unset).
if isinstance(param_realm_raw, list):
param_realms = [r.strip() for r in param_realm_raw if r and r.strip()]
elif isinstance(param_realm_raw, str) and "," in param_realm_raw:
param_realms = [r.strip() for r in param_realm_raw.split(",") if r.strip()]
elif isinstance(param_realm_raw, str) and param_realm_raw.strip():
param_realms = [param_realm_raw.strip()]
elif param_realm_raw:
param_realms = [param_realm_raw]
else:
param_realms = []
user_resolver = None
user_realm = None
if user is not None:
user_resolver = user.resolver
user_realm = user.realm
# Map each realm we will query to the resolver that should narrow it (or None
# for "every resolver in the realm"). A resolver narrows only the realm(s) it was
# supplied with: param_resolver narrows every realm in param_realms, user_resolver
# narrows only the user's realm. Realms are deduped; if a realm is named by both
# sources, a resolver from either source narrows it.
realm_filters = {}
for realm in param_realms:
realm_filters[realm] = param_resolver
if not realm_filters and param_resolver:
# A resolver was given without an explicit realm: seed its own scope (every
# realm containing it) before folding in the user's realm below, so a
# user object passed alongside a resolver-only query narrows/extends that
# scope instead of silently replacing it with just the user's own realm.
realm_filters = {realm: param_resolver for realm in get_realms_of_resolver(param_resolver)}
if user_realm:
realm_filters[user_realm] = realm_filters.get(user_realm) or user_resolver
if not (param_resolver or user_resolver or param_realm_raw or user_realm):
# Neither realm nor resolver was specified: search every resolver in all realms.
# Note: we test param_realm_raw (not the normalised param_realms) so that a
# non-empty realm filter which normalises to no valid realms (e.g. ",") does
# not fall through to searching every realm.
log.debug("Seldom event: Calling get_user_list with absolutely no information on realms or resolvers!")
realm_filters = {realm: None for realm in get_realms()}
if not realm_filters:
# A realm filter was given but normalised to no valid realm names (e.g. "," or
# " , "), or a resolver was given without a realm.
resolver_name = param_resolver or user_resolver
if not resolver_name:
return []
# No realm given but a resolver is given: query it in every realm that contains it.
realm_filters = {realm: resolver_name for realm in get_realms_of_resolver(resolver_name)}
if not realm_filters:
log.warning(f"Resolver '{resolver_name}' is not assigned to any realm.")
return []
# Determine some display values. Work on a local copy of requested_attributes
# so the caller's list is never mutated as a side effect of this call.
requested_attributes = list(requested_attributes) if requested_attributes is not None else None
remove_user_id = False
if include_custom_attributes and requested_attributes and "userid" not in requested_attributes:
# user id is required to later get the custom attributes for the user
requested_attributes.append("userid")
remove_user_id = True
# username is always required for deduplication across resolvers
remove_username = False
if requested_attributes and "username" not in requested_attributes:
remove_username = True
log.debug(f"With this search dictionary: {search_dict!r}")
requested_pi_user_attributes = list({"realm", "resolver", "editable"}.intersection(requested_attributes or []))
requested_user_store_attributes = list(set(requested_attributes or []) - set(requested_pi_user_attributes))
# Always fetch username from the resolver for dedup, even if not requested by the caller.
# If requested_attributes contains only PI attributes, requested_user_store_attributes would be empty, and some
# resolvers treat an empty list as "fetch all".
if requested_attributes and "username" not in requested_user_store_attributes:
requested_user_store_attributes.append("username")
succeeded_resolvers = set()
for realm, resolver_filter in realm_filters.items():
realm_config = get_realms(realm)
resolvers = get_ordered_resolvers(realm, realm_config=realm_config)
# A resolver assigned to this realm but pinned to a different node is filtered
# out of get_ordered_resolvers() before we ever see it, whether or not the
# caller asked for a specific resolver. Record it here so a plain realm (or
# all-realms) query does not silently report incomplete data as complete.
if failures is not None:
assigned_resolvers = {entry.get("name")
for entry in realm_config.get(realm, {}).get("resolver", [])}
for missing_name in assigned_resolvers - set(resolvers):
if missing_name not in failures:
failures.append(missing_name)
if resolver_filter:
if resolver_filter not in resolvers:
log.info(f"Resolver {resolver_filter!r} is not queryable in realm {realm!r}, skipping.")
continue
resolvers = [resolver_filter]
realm_id = get_realm_id(realm)
for resolver_name in resolvers:
try:
log.debug(f"Check for resolver class: {resolver_name!r}")
resolver = get_resolver_object(resolver_name)
# Continue if we couldn't find a resolver with the given name
if not resolver:
log.info(f"Can not find a resolver with the name '{resolver_name}'")
continue
# Pass a copy of the attribute list: some resolvers (e.g. SQLIdResolver) append to it in place,
# which would otherwise leak attributes like "userid" into the results of subsequent resolvers.
user_list = resolver.getUserList(search_dict, list(requested_user_store_attributes))
succeeded_resolvers.add(resolver_name)
for user_info in user_list:
if not requested_attributes or "realm" in requested_pi_user_attributes:
user_info["realm"] = realm
if not requested_attributes or "resolver" in requested_pi_user_attributes:
user_info["resolver"] = resolver_name
if not requested_attributes or "editable" in requested_pi_user_attributes:
user_info["editable"] = resolver.editable
if include_custom_attributes and realm_id is not None:
# Add the custom attributes, by class method from User
# with uid, resolvername and realm_id, which we need to determine by the realm name
custom_attributes = get_attributes(user_info.get("userid"), resolver_name, realm_id,
requested_attributes)
user_info.update(custom_attributes)
if remove_user_id:
# Remove the userid if it is not requested, as it is only needed for the custom attributes
user_info.pop("userid", None)
# Add user to users_dict, if it is not contained, yet.
# Deduplication across the resolvers of a realm relies on the username. This requires the
# resolver's user listing to actually return the username. SQL and LDAP resolvers always do,
# but for HTTP-based resolvers (e.g. Keycloak, Entra ID) it depends on the configured listing
# endpoint and attribute mapping. If the listing does not return a username, all users of that
# resolver share the same (None, realm) key - or ("", realm) if the mapped field is present but
# empty - and collapse into a single entry.
user_tuple = (user_info.get("username"), realm)
if remove_username:
user_info.pop("username", None)
if user_tuple not in users_dict:
users_dict[user_tuple] = user_info
log.debug(f"Found this userlist: {user_list!r}")
except (ResolverError, ParameterError) as ex:
# In case of wrong search parameters or broken resolver we continue.
# All other errors will be passed down. The skipped resolver name is
# recorded once in the optional ``failures`` list so callers can surface
# partial failures.
log.warning(f"Unable to get user list for resolver '{resolver_name}': {ex!r}")
log.debug(f"{traceback.format_exc()!s}")
if failures is not None and resolver_name not in failures:
failures.append(resolver_name)
continue
# A resolver can be queried in several realms (resolver-only queries iterate
# every realm containing it). If it failed in one realm but returned a user
# list in another, it is not a skipped resolver overall, so drop it: callers
# should only see resolvers that could not be queried anywhere.
if failures is not None and succeeded_resolvers:
failures[:] = [name for name in failures if name not in succeeded_resolvers]
users = list(users_dict.values())
return users
[docs]
@log_with(log)
@user_cache(cache_username)
def get_username(user_id: str, resolvername: str) -> str:
"""
Determine the username for a given id and a resolvername.
:param user_id: The id of the user in a resolver
:type user_id: string
:param resolvername: The name of the resolver
:return: the username or "" if it does not exist
:rtype: string
"""
username = ""
if user_id:
resolver = get_resolver_object(resolvername)
if resolver:
username = cached_username(resolver, resolvername, user_id)
return username
[docs]
def log_used_user(user: User, other_text: str = "") -> str:
"""
This creates a log message combined of a user and another text.
The user information is only added, if user.login != user.used_login
:param user: A user to log
:type user: User object
:param other_text: Some additional text
:return: str
"""
return f"logged in as {user.used_login}. {other_text}" if user.used_login != user.login else other_text
[docs]
def get_attributes(uid: str, resolver: str, realm_id: int, requested_attributes: list[str] = None) -> dict:
"""
Returns the attributes for the given user.
:param uid: The UID of the user
:param resolver: The name of the resolver
:param realm_id: The realm_id
:param requested_attributes: A list of attributes to return. If None, all attributes are returned.
:return: A dictionary of key/values
"""
stmt = select(CustomUserAttribute).filter_by(user_id=uid, resolver=resolver, realm_id=realm_id)
if requested_attributes:
stmt = stmt.filter(CustomUserAttribute.Key.in_(requested_attributes))
attributes = db.session.scalars(stmt).all()
custom_attributes = {attribute.Key: attribute.Value for attribute in attributes}
return custom_attributes
[docs]
def get_internal_attributes(uid: str, resolver: str, realm_id: int) -> dict:
"""
Returns all internal attributes for the given user as a single dict.
"""
stmt = select(InternalUserAttribute).filter_by(user_id=uid, resolver=resolver, realm_id=realm_id)
rows = db.session.scalars(stmt).all()
return {row.Key: row.Value for row in rows}
def _find_orphaned_attributes(model, orphaned_on_error: bool = False) -> list[tuple[str, str, int | None]]:
"""
Return the ``(user_id, resolver, realm_id)`` tuples in *model* whose user
can no longer be resolved.
Shared by the internal- and custom-attribute janitors. *model* must expose
``user_id``, ``resolver`` and ``realm_id`` columns (i.e.
:class:`InternalUserAttribute` or :class:`CustomUserAttribute`).
privacyIDEA does not own the user store, so deleting a user upstream leaves
behind attribute rows that can never be reached again through the normal
:class:`User` API. The token janitor calls this helper to find and prune
them.
:param model: the attribute model class to scan.
:param orphaned_on_error: If the resolver raises while looking up the user,
treat that row as orphaned. Mirrors the semantics of
:meth:`TokenClass.is_orphaned`.
The returned tuples may contain empty strings for ``user_id`` or
``resolver`` for legacy rows that predate the write-side guard against
empty identifiers — those rows are also reported as orphaned so the
janitor can prune them.
"""
from privacyidea.lib.resolver import get_resolver_object
stmt = select(model.user_id, model.resolver, model.realm_id).distinct()
rows = db.session.execute(stmt).all()
orphans: list[tuple[str, str, int | None]] = []
resolver_cache: dict[str, Any] = {}
for user_id, resolver_name, realm_id in rows:
if not user_id or not resolver_name:
# Rows with empty identifiers can never be reached through the
# User API, so we treat them as orphaned and let the janitor
# delete them.
orphans.append((user_id, resolver_name, realm_id))
continue
if resolver_name not in resolver_cache:
resolver_cache[resolver_name] = get_resolver_object(resolver_name)
resolver = resolver_cache[resolver_name]
if resolver is None:
# Resolver was deleted — everything under it is orphaned.
orphans.append((user_id, resolver_name, realm_id))
continue
try:
login = resolver.getUsername(user_id)
except Exception:
if orphaned_on_error:
orphans.append((user_id, resolver_name, realm_id))
continue
if not login:
# An empty username (without an exception) is the resolver's
# "this uid no longer exists" answer — the primary orphan signal,
# matching TokenClass.is_orphaned. Transient backend failures raise
# instead and are handled above via orphaned_on_error.
orphans.append((user_id, resolver_name, realm_id))
return orphans
def _delete_orphaned_attributes(model, orphans: list[tuple[str, str, int | None]]) -> int:
"""
Delete every *model* row whose ``(user_id, resolver, realm_id)`` matches one
of *orphans*. Returns the total number of rows deleted.
Bypasses the :class:`User` API on purpose — orphans are by definition users
that cannot be constructed.
"""
total = 0
for user_id, resolver_name, realm_id in orphans:
stmt = delete(model).filter_by(user_id=user_id, resolver=resolver_name, realm_id=realm_id)
total += db.session.execute(stmt).rowcount
db.session.commit()
return total
[docs]
def find_orphaned_internal_attributes(orphaned_on_error: bool = False) -> list[tuple[str, str, int | None]]:
"""
Return the ``(user_id, resolver, realm_id)`` tuples in ``internaluserattribute``
whose user is no longer in the resolver. See :func:`_find_orphaned_attributes`.
"""
return _find_orphaned_attributes(InternalUserAttribute, orphaned_on_error=orphaned_on_error)
[docs]
def delete_orphaned_internal_attributes(orphans: list[tuple[str, str, int | None]]) -> int:
"""
Delete every ``internaluserattribute`` row belonging to one of *orphans*.
See :func:`_delete_orphaned_attributes`.
"""
return _delete_orphaned_attributes(InternalUserAttribute, orphans)
[docs]
def find_orphaned_custom_attributes(orphaned_on_error: bool = False) -> list[tuple[str, str, int | None]]:
"""
Return the ``(user_id, resolver, realm_id)`` tuples in ``customuserattribute``
whose user is no longer in the resolver. See :func:`_find_orphaned_attributes`.
"""
return _find_orphaned_attributes(CustomUserAttribute, orphaned_on_error=orphaned_on_error)
[docs]
def delete_orphaned_custom_attributes(orphans: list[tuple[str, str, int | None]]) -> int:
"""
Delete every ``customuserattribute`` row belonging to one of *orphans*.
See :func:`_delete_orphaned_attributes`.
"""
return _delete_orphaned_attributes(CustomUserAttribute, orphans)
[docs]
def is_attribute_at_all() -> bool:
"""
Check if there are custom user attributes at all
"""
return bool(CustomUserAttribute.query.count())
