Mirror
/
zulip
mirror of https://github.com/zulip/zulip.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
zulip/zerver/models/users.py

1175 lines
45 KiB
Python
Raw Blame History

from email.headerregistry import Address
from typing import TYPE_CHECKING, Any, Optional
from uuid import uuid4
from django.conf import settings
from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin, UserManager
from django.db import models
from django.db.models import CASCADE, F, Q, QuerySet
from django.db.models.functions import Upper
from django.db.models.signals import post_save
from django.utils.timezone import now as timezone_now
from django.utils.translation import gettext_lazy
from typing_extensions import override
from zerver.lib.cache import (
active_non_guest_user_ids_cache_key,
active_user_ids_cache_key,
bot_dict_fields,
bot_dicts_in_realm_cache_key,
bot_profile_cache_key,
cache_with_key,
flush_user_profile,
realm_user_dict_fields,
realm_user_dicts_cache_key,
user_profile_by_api_key_cache_key,
user_profile_by_id_cache_key,
user_profile_cache_key,
)
from zerver.lib.types import ProfileData, RawUserDict
from zerver.lib.utils import generate_api_key
from zerver.models.constants import MAX_LANGUAGE_ID_LENGTH
if TYPE_CHECKING:
from zerver.models import Realm
class UserBaseSettings(models.Model):
"""This abstract class is the container for all preferences/personal
settings for users that control the behavior of the application.
It was extracted from UserProfile to support the RealmUserDefault
model (i.e. allow individual realms to configure the default
values of these preferences for new users in their organization).
Changing the default value for a field declared here likely
requires a migration to update all RealmUserDefault rows that had
the old default value to have the new default value. Otherwise,
the default change will only affect new users joining Realms
created after the change.
"""
### Generic UI settings
enter_sends = models.BooleanField(default=False)
### Preferences. ###
# left_side_userlist was removed from the UI in Zulip 6.0; the
# database model is being temporarily preserved in case we want to
# restore a version of the setting, preserving who had it enabled.
left_side_userlist = models.BooleanField(default=False)
default_language = models.CharField(default="en", max_length=MAX_LANGUAGE_ID_LENGTH)
# This setting controls which view is rendered first when Zulip loads.
# Values for it are URL suffix after `#`.
web_home_view = models.TextField(default="inbox")
web_escape_navigates_to_home_view = models.BooleanField(default=True)
fluid_layout_width = models.BooleanField(default=False)
high_contrast_mode = models.BooleanField(default=False)
translate_emoticons = models.BooleanField(default=False)
display_emoji_reaction_users = models.BooleanField(default=True)
twenty_four_hour_time = models.BooleanField(default=False)
starred_message_counts = models.BooleanField(default=True)
COLOR_SCHEME_AUTOMATIC = 1
COLOR_SCHEME_DARK = 2
COLOR_SCHEME_LIGHT = 3
COLOR_SCHEME_CHOICES = [COLOR_SCHEME_AUTOMATIC, COLOR_SCHEME_DARK, COLOR_SCHEME_LIGHT]
color_scheme = models.PositiveSmallIntegerField(default=COLOR_SCHEME_AUTOMATIC)
# Information density is established through
# adjustments to the font size and line height.
WEB_FONT_SIZE_PX_COMPACT = 14
WEB_FONT_SIZE_PX_DEFAULT = 16
WEB_LINE_HEIGHT_PERCENT_COMPACT = 122
WEB_LINE_HEIGHT_PERCENT_DEFAULT = 140
dense_mode = models.BooleanField(default=False)
web_font_size_px = models.PositiveSmallIntegerField(default=WEB_FONT_SIZE_PX_DEFAULT)
web_line_height_percent = models.PositiveSmallIntegerField(
default=WEB_LINE_HEIGHT_PERCENT_DEFAULT
)
# UI setting to control how animated images are played.
web_animate_image_previews = models.TextField(default="on_hover")
# UI setting controlling Zulip's behavior of demoting in the sort
# order and graying out streams with no recent traffic. The
# default behavior, automatic, enables this behavior once a user
# is subscribed to 30+ streams in the web app.
DEMOTE_STREAMS_AUTOMATIC = 1
DEMOTE_STREAMS_ALWAYS = 2
DEMOTE_STREAMS_NEVER = 3
DEMOTE_STREAMS_CHOICES = [
DEMOTE_STREAMS_AUTOMATIC,
DEMOTE_STREAMS_ALWAYS,
DEMOTE_STREAMS_NEVER,
]
demote_inactive_streams = models.PositiveSmallIntegerField(default=DEMOTE_STREAMS_AUTOMATIC)
# UI setting controlling whether or not the Zulip web app will
# mark messages as read as it scrolls through the feed.
MARK_READ_ON_SCROLL_ALWAYS = 1
MARK_READ_ON_SCROLL_CONVERSATION_ONLY = 2
MARK_READ_ON_SCROLL_NEVER = 3
WEB_MARK_READ_ON_SCROLL_POLICY_CHOICES = [
MARK_READ_ON_SCROLL_ALWAYS,
MARK_READ_ON_SCROLL_CONVERSATION_ONLY,
MARK_READ_ON_SCROLL_NEVER,
]
web_mark_read_on_scroll_policy = models.SmallIntegerField(default=MARK_READ_ON_SCROLL_ALWAYS)
# UI setting controlling if clicking on a channel link should open
# the channel feed (interleaved view) or narrow to the first topic
# in the channel.
WEB_CHANNEL_DEFAULT_VIEW_FIRST_TOPIC = 1
WEB_CHANNEL_DEFAULT_VIEW_CHANNEL_FEED = 2
WEB_CHANNEL_DEFAULT_VIEW_CHOICES = [
WEB_CHANNEL_DEFAULT_VIEW_FIRST_TOPIC,
WEB_CHANNEL_DEFAULT_VIEW_CHANNEL_FEED,
]
web_channel_default_view = models.SmallIntegerField(
default=WEB_CHANNEL_DEFAULT_VIEW_FIRST_TOPIC,
db_default=WEB_CHANNEL_DEFAULT_VIEW_FIRST_TOPIC,
)
# Emoji sets
GOOGLE_EMOJISET = "google"
GOOGLE_BLOB_EMOJISET = "google-blob"
TEXT_EMOJISET = "text"
TWITTER_EMOJISET = "twitter"
EMOJISET_CHOICES = (
(GOOGLE_EMOJISET, "Google"),
(TWITTER_EMOJISET, "Twitter"),
(TEXT_EMOJISET, "Plain text"),
(GOOGLE_BLOB_EMOJISET, "Google blobs"),
)
emojiset = models.CharField(default=GOOGLE_EMOJISET, choices=EMOJISET_CHOICES, max_length=20)
# User list style
USER_LIST_STYLE_COMPACT = 1
USER_LIST_STYLE_WITH_STATUS = 2
USER_LIST_STYLE_WITH_AVATAR = 3
USER_LIST_STYLE_CHOICES = [
USER_LIST_STYLE_COMPACT,
USER_LIST_STYLE_WITH_STATUS,
USER_LIST_STYLE_WITH_AVATAR,
]
user_list_style = models.PositiveSmallIntegerField(default=USER_LIST_STYLE_WITH_STATUS)
# Show unread counts for
WEB_STREAM_UNREADS_COUNT_DISPLAY_POLICY_ALL_STREAMS = 1
WEB_STREAM_UNREADS_COUNT_DISPLAY_POLICY_UNMUTED_STREAMS = 2
WEB_STREAM_UNREADS_COUNT_DISPLAY_POLICY_NO_STREAMS = 3
WEB_STREAM_UNREADS_COUNT_DISPLAY_POLICY_CHOICES = [
WEB_STREAM_UNREADS_COUNT_DISPLAY_POLICY_ALL_STREAMS,
WEB_STREAM_UNREADS_COUNT_DISPLAY_POLICY_UNMUTED_STREAMS,
WEB_STREAM_UNREADS_COUNT_DISPLAY_POLICY_NO_STREAMS,
]
web_stream_unreads_count_display_policy = models.PositiveSmallIntegerField(
default=WEB_STREAM_UNREADS_COUNT_DISPLAY_POLICY_UNMUTED_STREAMS
)
# Setting to control whether to automatically go to the
# conversation where message was sent.
web_navigate_to_sent_message = models.BooleanField(default=True)
### Notifications settings. ###
email_notifications_batching_period_seconds = models.IntegerField(default=120)
# Stream notifications.
enable_stream_desktop_notifications = models.BooleanField(default=False)
enable_stream_email_notifications = models.BooleanField(default=False)
enable_stream_push_notifications = models.BooleanField(default=False)
enable_stream_audible_notifications = models.BooleanField(default=False)
notification_sound = models.CharField(max_length=20, default="zulip")
wildcard_mentions_notify = models.BooleanField(default=True)
# Followed Topics notifications.
enable_followed_topic_desktop_notifications = models.BooleanField(default=True)
enable_followed_topic_email_notifications = models.BooleanField(default=True)
enable_followed_topic_push_notifications = models.BooleanField(default=True)
enable_followed_topic_audible_notifications = models.BooleanField(default=True)
enable_followed_topic_wildcard_mentions_notify = models.BooleanField(default=True)
# Direct message + @-mention notifications.
enable_desktop_notifications = models.BooleanField(default=True)
pm_content_in_desktop_notifications = models.BooleanField(default=True)
enable_sounds = models.BooleanField(default=True)
enable_offline_email_notifications = models.BooleanField(default=True)
message_content_in_email_notifications = models.BooleanField(default=True)
enable_offline_push_notifications = models.BooleanField(default=True)
enable_online_push_notifications = models.BooleanField(default=True)
DESKTOP_ICON_COUNT_DISPLAY_MESSAGES = 1
DESKTOP_ICON_COUNT_DISPLAY_DM_MENTION_FOLLOWED_TOPIC = 2
DESKTOP_ICON_COUNT_DISPLAY_DM_MENTION = 3
DESKTOP_ICON_COUNT_DISPLAY_NONE = 4
DESKTOP_ICON_COUNT_DISPLAY_CHOICES = [
DESKTOP_ICON_COUNT_DISPLAY_MESSAGES,
DESKTOP_ICON_COUNT_DISPLAY_DM_MENTION,
DESKTOP_ICON_COUNT_DISPLAY_DM_MENTION_FOLLOWED_TOPIC,
DESKTOP_ICON_COUNT_DISPLAY_NONE,
]
desktop_icon_count_display = models.PositiveSmallIntegerField(
default=DESKTOP_ICON_COUNT_DISPLAY_MESSAGES
)
enable_digest_emails = models.BooleanField(default=True)
enable_login_emails = models.BooleanField(default=True)
enable_marketing_emails = models.BooleanField(default=True)
presence_enabled = models.BooleanField(default=True)
REALM_NAME_IN_EMAIL_NOTIFICATIONS_POLICY_AUTOMATIC = 1
REALM_NAME_IN_EMAIL_NOTIFICATIONS_POLICY_ALWAYS = 2
REALM_NAME_IN_EMAIL_NOTIFICATIONS_POLICY_NEVER = 3
REALM_NAME_IN_EMAIL_NOTIFICATIONS_POLICY_CHOICES = [
REALM_NAME_IN_EMAIL_NOTIFICATIONS_POLICY_AUTOMATIC,
REALM_NAME_IN_EMAIL_NOTIFICATIONS_POLICY_ALWAYS,
REALM_NAME_IN_EMAIL_NOTIFICATIONS_POLICY_NEVER,
]
realm_name_in_email_notifications_policy = models.PositiveSmallIntegerField(
default=REALM_NAME_IN_EMAIL_NOTIFICATIONS_POLICY_AUTOMATIC
)
# The following two settings control which topics to automatically
# 'follow' or 'unmute in a muted stream', respectively.
# Follow or unmute a topic automatically on:
# - PARTICIPATION: Send a message, React to a message, Participate in a poll or Edit a TO-DO list.
# - SEND: Send a message.
# - INITIATION: Send the first message in the topic.
# - NEVER: Never automatically follow or unmute a topic.
AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION = 1
AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_SEND = 2
AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION = 3
AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_NEVER = 4
AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_CHOICES = [
AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_SEND,
AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_NEVER,
]
automatically_follow_topics_policy = models.PositiveSmallIntegerField(
default=AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
)
automatically_unmute_topics_in_muted_streams_policy = models.PositiveSmallIntegerField(
default=AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_SEND,
)
automatically_follow_topics_where_mentioned = models.BooleanField(default=True)
# Whether or not the user wants to sync their drafts.
enable_drafts_synchronization = models.BooleanField(default=True)
# Privacy settings
send_stream_typing_notifications = models.BooleanField(default=True)
send_private_typing_notifications = models.BooleanField(default=True)
send_read_receipts = models.BooleanField(default=True)
allow_private_data_export = models.BooleanField(default=False)
# Whether the user wants to see typing notifications.
receives_typing_notifications = models.BooleanField(default=True)
# Who in the organization has access to users' actual email
# addresses. Controls whether the UserProfile.email field is
# the same as UserProfile.delivery_email, or is instead a fake
# generated value encoding the user ID and realm hostname.
EMAIL_ADDRESS_VISIBILITY_EVERYONE = 1
EMAIL_ADDRESS_VISIBILITY_MEMBERS = 2
EMAIL_ADDRESS_VISIBILITY_ADMINS = 3
EMAIL_ADDRESS_VISIBILITY_NOBODY = 4
EMAIL_ADDRESS_VISIBILITY_MODERATORS = 5
email_address_visibility = models.PositiveSmallIntegerField(
default=EMAIL_ADDRESS_VISIBILITY_EVERYONE,
)
EMAIL_ADDRESS_VISIBILITY_ID_TO_NAME_MAP = {
EMAIL_ADDRESS_VISIBILITY_EVERYONE: gettext_lazy("Admins, moderators, members and guests"),
EMAIL_ADDRESS_VISIBILITY_MEMBERS: gettext_lazy("Admins, moderators and members"),
EMAIL_ADDRESS_VISIBILITY_MODERATORS: gettext_lazy("Admins and moderators"),
EMAIL_ADDRESS_VISIBILITY_ADMINS: gettext_lazy("Admins only"),
EMAIL_ADDRESS_VISIBILITY_NOBODY: gettext_lazy("Nobody"),
}
EMAIL_ADDRESS_VISIBILITY_TYPES = list(EMAIL_ADDRESS_VISIBILITY_ID_TO_NAME_MAP.keys())
display_settings_legacy = dict(
# Don't add anything new to this legacy dict.
# Instead, see `modern_settings` below.
color_scheme=int,
default_language=str,
web_home_view=str,
demote_inactive_streams=int,
dense_mode=bool,
emojiset=str,
enable_drafts_synchronization=bool,
enter_sends=bool,
fluid_layout_width=bool,
high_contrast_mode=bool,
left_side_userlist=bool,
starred_message_counts=bool,
translate_emoticons=bool,
twenty_four_hour_time=bool,
)
notification_settings_legacy = dict(
# Don't add anything new to this legacy dict.
# Instead, see `modern_notification_settings` below.
desktop_icon_count_display=int,
email_notifications_batching_period_seconds=int,
enable_desktop_notifications=bool,
enable_digest_emails=bool,
enable_login_emails=bool,
enable_marketing_emails=bool,
enable_offline_email_notifications=bool,
enable_offline_push_notifications=bool,
enable_online_push_notifications=bool,
enable_sounds=bool,
enable_stream_audible_notifications=bool,
enable_stream_desktop_notifications=bool,
enable_stream_email_notifications=bool,
enable_stream_push_notifications=bool,
message_content_in_email_notifications=bool,
notification_sound=str,
pm_content_in_desktop_notifications=bool,
presence_enabled=bool,
realm_name_in_email_notifications_policy=int,
wildcard_mentions_notify=bool,
)
modern_settings = dict(
# Add new general settings here.
display_emoji_reaction_users=bool,
email_address_visibility=int,
web_escape_navigates_to_home_view=bool,
receives_typing_notifications=bool,
send_private_typing_notifications=bool,
send_read_receipts=bool,
send_stream_typing_notifications=bool,
allow_private_data_export=bool,
web_mark_read_on_scroll_policy=int,
web_channel_default_view=int,
user_list_style=int,
web_animate_image_previews=str,
web_stream_unreads_count_display_policy=int,
web_font_size_px=int,
web_line_height_percent=int,
web_navigate_to_sent_message=bool,
)
modern_notification_settings: dict[str, Any] = dict(
# Add new notification settings here.
enable_followed_topic_desktop_notifications=bool,
enable_followed_topic_email_notifications=bool,
enable_followed_topic_push_notifications=bool,
enable_followed_topic_audible_notifications=bool,
enable_followed_topic_wildcard_mentions_notify=bool,
automatically_follow_topics_policy=int,
automatically_unmute_topics_in_muted_streams_policy=int,
automatically_follow_topics_where_mentioned=bool,
)
notification_setting_types = {
**notification_settings_legacy,
**modern_notification_settings,
}
# Define the types of the various automatically managed properties
property_types = {
**display_settings_legacy,
**notification_setting_types,
**modern_settings,
}
class Meta:
abstract = True
@staticmethod
def emojiset_choices() -> list[dict[str, str]]:
return [
dict(key=emojiset[0], text=emojiset[1]) for emojiset in UserProfile.EMOJISET_CHOICES
]
class RealmUserDefault(UserBaseSettings):
"""This table stores realm-level default values for user preferences
like notification settings, used when creating a new user account.
"""
realm = models.OneToOneField("zerver.Realm", on_delete=CASCADE)
class UserProfile(AbstractBaseUser, PermissionsMixin, UserBaseSettings):
USERNAME_FIELD = "email"
MAX_NAME_LENGTH = 100
MIN_NAME_LENGTH = 2
API_KEY_LENGTH = 32
NAME_INVALID_CHARS = ["*", "`", "\\", ">", '"', "@"]
DEFAULT_BOT = 1
"""
Incoming webhook bots are limited to only sending messages via webhooks.
Thus, it is less of a security risk to expose their API keys to third-party services,
since they can't be used to read messages.
"""
INCOMING_WEBHOOK_BOT = 2
# This value is also being used in web/src/settings_bots.js.
# On updating it here, update it there as well.
OUTGOING_WEBHOOK_BOT = 3
"""
Embedded bots run within the Zulip server itself; events are added to the
embedded_bots queue and then handled by a QueueProcessingWorker.
"""
EMBEDDED_BOT = 4
BOT_TYPES = {
DEFAULT_BOT: "Generic bot",
INCOMING_WEBHOOK_BOT: "Incoming webhook",
OUTGOING_WEBHOOK_BOT: "Outgoing webhook",
EMBEDDED_BOT: "Embedded bot",
}
SERVICE_BOT_TYPES = [
OUTGOING_WEBHOOK_BOT,
EMBEDDED_BOT,
]
id = models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")
# For historical reasons, Zulip has two email fields. The
# `delivery_email` field is the user's email address, where all
# email notifications will be sent, and is used for all
# authentication use cases.
#
# The `email` field is the same as delivery_email in organizations
# with EMAIL_ADDRESS_VISIBILITY_EVERYONE. For other
# organizations, it will be a unique value of the form
# user1234@example.com. This field exists for backwards
# compatibility in Zulip APIs where users are referred to by their
# email address, not their ID; it should be used in all API use cases.
#
# Both fields are unique within a realm (in a case-insensitive
# fashion). Since Django's unique_together is case sensitive, this
# is enforced via SQL indexes created by
# zerver/migrations/0295_case_insensitive_email_indexes.py.
delivery_email = models.EmailField(blank=False, db_index=True)
email = models.EmailField(blank=False, db_index=True)
realm = models.ForeignKey("zerver.Realm", on_delete=CASCADE)
# Foreign key to the Recipient object for PERSONAL type messages to this user.
recipient = models.ForeignKey("zerver.Recipient", null=True, on_delete=models.SET_NULL)
INACCESSIBLE_USER_NAME = gettext_lazy("Unknown user")
# The user's name. We prefer the model of a full_name
# over first+last because cultures vary on how many
# names one has, whether the family name is first or last, etc.
# It also allows organizations to encode a bit of non-name data in
# the "name" attribute if desired, like gender pronouns,
# graduation year, etc.
full_name = models.CharField(max_length=MAX_NAME_LENGTH)
date_joined = models.DateTimeField(default=timezone_now)
# Terms of Service version number that this user has accepted. We
# use the special value TOS_VERSION_BEFORE_FIRST_LOGIN for users
# whose account was created without direct user interaction (via
# the API or a data import), and null for users whose account is
# fully created on servers that do not have a configured ToS.
TOS_VERSION_BEFORE_FIRST_LOGIN = "-1"
tos_version = models.CharField(null=True, max_length=10)
api_key = models.CharField(max_length=API_KEY_LENGTH, default=generate_api_key, unique=True)
# A UUID generated on user creation. Introduced primarily to
# provide a unique key for a user for the mobile push
# notifications bouncer that will not have collisions after doing
# a data export and then import.
uuid = models.UUIDField(default=uuid4, unique=True)
# Whether the user has access to server-level administrator pages, like /activity
is_staff = models.BooleanField(default=False)
# For a normal user, this is True unless the user or an admin has
# deactivated their account. The name comes from Django; this field
# isn't related to presence or to whether the user has recently used Zulip.
#
# See also `long_term_idle`.
is_active = models.BooleanField(default=True, db_index=True)
is_billing_admin = models.BooleanField(default=False, db_index=True)
is_bot = models.BooleanField(default=False, db_index=True)
bot_type = models.PositiveSmallIntegerField(null=True, db_index=True)
bot_owner = models.ForeignKey("self", null=True, on_delete=models.SET_NULL)
# Each role has a superset of the permissions of the next higher
# numbered role. When adding new roles, leave enough space for
# future roles to be inserted between currently adjacent
# roles. These constants appear in RealmAuditLog.extra_data, so
# changes to them will require a migration of RealmAuditLog.
ROLE_REALM_OWNER = 100
ROLE_REALM_ADMINISTRATOR = 200
ROLE_MODERATOR = 300
ROLE_MEMBER = 400
ROLE_GUEST = 600
role = models.PositiveSmallIntegerField(default=ROLE_MEMBER, db_index=True)
ROLE_TYPES = [
ROLE_REALM_OWNER,
ROLE_REALM_ADMINISTRATOR,
ROLE_MODERATOR,
ROLE_MEMBER,
ROLE_GUEST,
]
# Whether the user has been "soft-deactivated" due to weeks of inactivity.
# For these users we avoid doing UserMessage table work, as an optimization
# for large Zulip organizations with lots of single-visit users.
long_term_idle = models.BooleanField(default=False, db_index=True)
# When we last added basic UserMessage rows for a long_term_idle user.
last_active_message_id = models.IntegerField(null=True)
# Mirror dummies are fake (!is_active) users used to provide
# message senders in our cross-protocol Zephyr<->Zulip content
# mirroring integration, so that we can display mirrored content
# like native Zulip messages (with a name + avatar, etc.).
is_mirror_dummy = models.BooleanField(default=False)
# Users with this flag set are allowed to forge messages as sent by another
# user and to send to private streams; also used for Zephyr/Jabber mirroring.
can_forge_sender = models.BooleanField(default=False, db_index=True)
# Users with this flag set can create other users via API.
can_create_users = models.BooleanField(default=False, db_index=True)
# Used for rate-limiting certain automated messages generated by bots
last_reminder = models.DateTimeField(default=None, null=True)
# Minutes to wait before warning a bot owner that their bot sent a message
# to a nonexistent stream
BOT_OWNER_STREAM_ALERT_WAITPERIOD = 1
# API rate limits, formatted as a comma-separated list of range:max pairs
rate_limits = models.CharField(default="", max_length=100)
# Default streams for some deprecated/legacy classes of bot users.
default_sending_stream = models.ForeignKey(
"zerver.Stream",
null=True,
related_name="+",
on_delete=models.SET_NULL,
)
default_events_register_stream = models.ForeignKey(
"zerver.Stream",
null=True,
related_name="+",
on_delete=models.SET_NULL,
)
default_all_public_streams = models.BooleanField(default=False)
# A time zone name from the `tzdata` database, as found in zoneinfo.available_timezones().
#
# The longest existing name is 32 characters long, so max_length=40 seems
# like a safe choice.
#
# In Django, the convention is to use an empty string instead of NULL/None
# for text-based fields. For more information, see
# https://docs.djangoproject.com/en/5.0/ref/models/fields/#django.db.models.Field.null.
timezone = models.CharField(max_length=40, default="")
AVATAR_FROM_GRAVATAR = "G"
AVATAR_FROM_USER = "U"
AVATAR_SOURCES = (
(AVATAR_FROM_GRAVATAR, "Hosted by Gravatar"),
(AVATAR_FROM_USER, "Uploaded by user"),
)
avatar_source = models.CharField(
default=AVATAR_FROM_GRAVATAR, choices=AVATAR_SOURCES, max_length=1
)
avatar_version = models.PositiveSmallIntegerField(default=1)
# This is only used for LDAP-provided avatars; it contains the
# SHA256 hex digest of most recent raw contents that LDAP provided
# us, pre-thumbnailing.
avatar_hash = models.CharField(null=True, max_length=64)
zoom_token = models.JSONField(default=None, null=True)
objects = UserManager()
ROLE_ID_TO_NAME_MAP = {
ROLE_REALM_OWNER: gettext_lazy("Organization owner"),
ROLE_REALM_ADMINISTRATOR: gettext_lazy("Organization administrator"),
ROLE_MODERATOR: gettext_lazy("Moderator"),
ROLE_MEMBER: gettext_lazy("Member"),
ROLE_GUEST: gettext_lazy("Guest"),
}
# Mapping of role ids to simple string identifiers for the roles,
# to be used in API contexts such as SCIM provisioning.
ROLE_ID_TO_API_NAME = {
ROLE_REALM_OWNER: "owner",
ROLE_REALM_ADMINISTRATOR: "administrator",
ROLE_MODERATOR: "moderator",
ROLE_MEMBER: "member",
ROLE_GUEST: "guest",
}
ROLE_API_NAME_TO_ID = {v: k for k, v in ROLE_ID_TO_API_NAME.items()}
class Meta:
constraints = [
models.UniqueConstraint(
"realm",
Upper(F("email")),
name="zerver_userprofile_realm_id_email_uniq",
),
models.UniqueConstraint(
"realm",
Upper(F("delivery_email")),
name="zerver_userprofile_realm_id_delivery_email_uniq",
),
]
indexes = [
models.Index(Upper("email"), name="upper_userprofile_email_idx"),
]
@override
def __str__(self) -> str:
return f"{self.email} {self.realm!r}"
def get_role_name(self) -> str:
return str(self.ROLE_ID_TO_NAME_MAP[self.role])
def profile_data(self) -> ProfileData:
from zerver.models import CustomProfileFieldValue
from zerver.models.custom_profile_fields import custom_profile_fields_for_realm
values = CustomProfileFieldValue.objects.filter(user_profile=self)
user_data = {
v.field_id: {"value": v.value, "rendered_value": v.rendered_value} for v in values
}
data: ProfileData = []
for field in custom_profile_fields_for_realm(self.realm_id):
field_values = user_data.get(field.id, None)
if field_values:
value, rendered_value = (
field_values.get("value"),
field_values.get("rendered_value"),
)
else:
value, rendered_value = None, None
field_type = field.field_type
if value is not None:
converter = field.FIELD_CONVERTERS[field_type]
value = converter(value)
field_data = field.as_dict()
data.append(
{
"id": field_data["id"],
"name": field_data["name"],
"type": field_data["type"],
"hint": field_data["hint"],
"field_data": field_data["field_data"],
"order": field_data["order"],
"value": value,
"rendered_value": rendered_value,
}
)
return data
def can_admin_user(self, target_user: "UserProfile") -> bool:
"""Returns whether this user has permission to modify target_user"""
if target_user.bot_owner_id == self.id:
return True
elif self.is_realm_admin and self.realm == target_user.realm:
return True
else:
return False
@property
def is_provisional_member(self) -> bool:
if self.is_moderator:
return False
diff = (timezone_now() - self.date_joined).days
if diff < self.realm.waiting_period_threshold:
return True
return False
@property
def is_realm_admin(self) -> bool:
return self.role in (UserProfile.ROLE_REALM_ADMINISTRATOR, UserProfile.ROLE_REALM_OWNER)
@is_realm_admin.setter
def is_realm_admin(self, value: bool) -> None:
if value:
self.role = UserProfile.ROLE_REALM_ADMINISTRATOR
elif self.role == UserProfile.ROLE_REALM_ADMINISTRATOR:
# We need to be careful to not accidentally change
# ROLE_GUEST to ROLE_MEMBER here.
self.role = UserProfile.ROLE_MEMBER
@property
def has_billing_access(self) -> bool:
return self.is_realm_owner or self.is_billing_admin
@property
def is_realm_owner(self) -> bool:
return self.role == UserProfile.ROLE_REALM_OWNER
@is_realm_owner.setter
def is_realm_owner(self, value: bool) -> None:
if value:
self.role = UserProfile.ROLE_REALM_OWNER
elif self.role == UserProfile.ROLE_REALM_OWNER:
# We need to be careful to not accidentally change
# ROLE_GUEST to ROLE_MEMBER here.
self.role = UserProfile.ROLE_MEMBER
@property
def is_guest(self) -> bool:
return self.role == UserProfile.ROLE_GUEST
@is_guest.setter
def is_guest(self, value: bool) -> None:
if value:
self.role = UserProfile.ROLE_GUEST
elif self.role == UserProfile.ROLE_GUEST:
# We need to be careful to not accidentally change
# ROLE_REALM_ADMINISTRATOR to ROLE_MEMBER here.
self.role = UserProfile.ROLE_MEMBER
@property
def is_moderator(self) -> bool:
return self.role == UserProfile.ROLE_MODERATOR
@is_moderator.setter
def is_moderator(self, value: bool) -> None:
if value:
self.role = UserProfile.ROLE_MODERATOR
elif self.role == UserProfile.ROLE_MODERATOR:
# We need to be careful to not accidentally change
# ROLE_GUEST to ROLE_MEMBER here.
self.role = UserProfile.ROLE_MEMBER
@property
def is_incoming_webhook(self) -> bool:
return self.bot_type == UserProfile.INCOMING_WEBHOOK_BOT
@property
def allowed_bot_types(self) -> list[int]:
from zerver.models.realms import BotCreationPolicyEnum
allowed_bot_types = []
if (
self.is_realm_admin
or self.realm.bot_creation_policy != BotCreationPolicyEnum.LIMIT_GENERIC_BOTS
):
allowed_bot_types.append(UserProfile.DEFAULT_BOT)
allowed_bot_types += [
UserProfile.INCOMING_WEBHOOK_BOT,
UserProfile.OUTGOING_WEBHOOK_BOT,
]
if settings.EMBEDDED_BOTS_ENABLED:
allowed_bot_types.append(UserProfile.EMBEDDED_BOT)
return allowed_bot_types
def email_address_is_realm_public(self) -> bool:
# Bots always have EMAIL_ADDRESS_VISIBILITY_EVERYONE.
if self.email_address_visibility == UserProfile.EMAIL_ADDRESS_VISIBILITY_EVERYONE:
return True
return False
def has_permission(self, policy_name: str, realm: Optional["Realm"] = None) -> bool:
from zerver.lib.user_groups import user_has_permission_for_group_setting
from zerver.models import Realm
if policy_name not in [
"add_custom_emoji_policy",
"can_create_groups",
"can_create_private_channel_group",
"can_create_public_channel_group",
"can_create_web_public_channel_group",
"can_delete_any_message_group",
"can_delete_own_message_group",
"can_manage_all_groups",
"create_multiuse_invite_group",
"direct_message_initiator_group",
"direct_message_permission_group",
"edit_topic_policy",
"invite_to_stream_policy",
"invite_to_realm_policy",
"move_messages_between_streams_policy",
]:
raise AssertionError("Invalid policy")
if policy_name in Realm.REALM_PERMISSION_GROUP_SETTINGS:
if realm is None:
# realm is passed by the caller only when we optimize
# the number of database queries by fetching the group
# setting fields using select_related.
realm = self.realm
allowed_user_group = getattr(realm, policy_name)
setting_config = Realm.REALM_PERMISSION_GROUP_SETTINGS[policy_name]
return user_has_permission_for_group_setting(allowed_user_group, self, setting_config)
policy_value = getattr(self.realm, policy_name)
if policy_value == Realm.POLICY_NOBODY:
return False
if policy_value == Realm.POLICY_EVERYONE:
return True
if self.is_realm_owner:
return True
if policy_value == Realm.POLICY_OWNERS_ONLY:
return False
if self.is_realm_admin:
return True
if policy_value == Realm.POLICY_ADMINS_ONLY:
return False
if self.is_moderator:
return True
if policy_value == Realm.POLICY_MODERATORS_ONLY:
return False
if self.is_guest:
return False
if policy_value == Realm.POLICY_MEMBERS_ONLY:
return True
assert policy_value == Realm.POLICY_FULL_MEMBERS_ONLY
return not self.is_provisional_member
def can_create_public_streams(self, realm: Optional["Realm"] = None) -> bool:
return self.has_permission("can_create_public_channel_group", realm)
def can_create_private_streams(self, realm: Optional["Realm"] = None) -> bool:
return self.has_permission("can_create_private_channel_group", realm)
def can_create_web_public_streams(self) -> bool:
if not self.realm.web_public_streams_enabled():
return False
return self.has_permission("can_create_web_public_channel_group")
def can_subscribe_other_users(self) -> bool:
return self.has_permission("invite_to_stream_policy")
def can_invite_users_by_email(self) -> bool:
return self.has_permission("invite_to_realm_policy")
def can_create_multiuse_invite_to_realm(self) -> bool:
return self.has_permission("create_multiuse_invite_group")
def can_move_messages_between_streams(self) -> bool:
return self.has_permission("move_messages_between_streams_policy")
def can_create_user_groups(self) -> bool:
return self.has_permission("can_create_groups")
def can_manage_all_groups(self) -> bool:
return self.has_permission("can_manage_all_groups")
def can_move_messages_to_another_topic(self) -> bool:
return self.has_permission("edit_topic_policy")
def can_add_custom_emoji(self) -> bool:
return self.has_permission("add_custom_emoji_policy")
def can_delete_any_message(self) -> bool:
return self.has_permission("can_delete_any_message_group")
def can_delete_own_message(self) -> bool:
return self.has_permission("can_delete_own_message_group")
def can_access_public_streams(self) -> bool:
return not (self.is_guest or self.realm.is_zephyr_mirror_realm)
def major_tos_version(self) -> int:
if self.tos_version is not None:
return int(self.tos_version.split(".")[0])
else:
return -1
def format_requester_for_logs(self) -> str:
return "{}@{}".format(self.id, self.realm.string_id or "root")
@override
def set_password(self, password: str | None) -> None:
if password is None:
self.set_unusable_password()
return
from zproject.backends import check_password_strength
if not check_password_strength(password):
raise PasswordTooWeakError
super().set_password(password)
class PasswordTooWeakError(Exception):
pass
def remote_user_to_email(remote_user: str) -> str:
if settings.SSO_APPEND_DOMAIN is not None:
return Address(username=remote_user, domain=settings.SSO_APPEND_DOMAIN).addr_spec
return remote_user
# Make sure we flush the UserProfile object from our remote cache
# whenever we save it.
post_save.connect(flush_user_profile, sender=UserProfile)
@cache_with_key(user_profile_by_id_cache_key, timeout=3600 * 24 * 7)
def get_user_profile_by_id(user_profile_id: int) -> UserProfile:
return UserProfile.objects.select_related(
"realm",
"realm__can_access_all_users_group",
"realm__can_access_all_users_group__named_user_group",
"realm__direct_message_initiator_group",
"realm__direct_message_initiator_group__named_user_group",
"realm__direct_message_permission_group",
"realm__direct_message_permission_group__named_user_group",
"bot_owner",
).get(id=user_profile_id)
def get_user_profile_by_email(email: str) -> UserProfile:
"""This function is intended to be used for
manual manage.py shell work; robust code must use get_user or
get_user_by_delivery_email instead, because Zulip supports
multiple users with a given (delivery) email address existing on a
single server (in different realms).
"""
return UserProfile.objects.select_related("realm").get(delivery_email__iexact=email.strip())
@cache_with_key(user_profile_by_api_key_cache_key, timeout=3600 * 24 * 7)
def maybe_get_user_profile_by_api_key(api_key: str) -> UserProfile | None:
try:
return UserProfile.objects.select_related(
"realm",
"realm__can_access_all_users_group",
"realm__can_access_all_users_group__named_user_group",
"bot_owner",
).get(api_key=api_key)
except UserProfile.DoesNotExist:
# We will cache failed lookups with None. The
# use case here is that broken API clients may
# continually ask for the same wrong API key, and
# we want to handle that as quickly as possible.
return None
def get_user_profile_by_api_key(api_key: str) -> UserProfile:
user_profile = maybe_get_user_profile_by_api_key(api_key)
if user_profile is None:
raise UserProfile.DoesNotExist
return user_profile
def get_user_by_delivery_email(email: str, realm: "Realm") -> UserProfile:
"""Fetches a user given their delivery email. For use in
authentication/registration contexts. Do not use for user-facing
views (e.g. Zulip API endpoints) as doing so would violate the
EMAIL_ADDRESS_VISIBILITY_ADMINS security model. Use get_user in
those code paths.
"""
return UserProfile.objects.select_related(
"realm",
"realm__can_access_all_users_group",
"realm__can_access_all_users_group__named_user_group",
"realm__direct_message_initiator_group",
"realm__direct_message_initiator_group__named_user_group",
"realm__direct_message_permission_group",
"realm__direct_message_permission_group__named_user_group",
"bot_owner",
).get(delivery_email__iexact=email.strip(), realm=realm)
def get_users_by_delivery_email(emails: set[str], realm: "Realm") -> QuerySet[UserProfile]:
"""This is similar to get_user_by_delivery_email, and
it has the same security caveats. It gets multiple
users and returns a QuerySet, since most callers
will only need two or three fields.
If you are using this to get large UserProfile objects, you are
probably making a mistake, but if you must,
then use `select_related`.
"""
"""
Django doesn't support delivery_email__iexact__in, so
we simply OR all the filters that we'd do for the
one-email case.
"""
email_filter = Q()
for email in emails:
email_filter |= Q(delivery_email__iexact=email.strip())
return UserProfile.objects.filter(realm=realm).filter(email_filter)
@cache_with_key(user_profile_cache_key, timeout=3600 * 24 * 7)
def get_user(email: str, realm: "Realm") -> UserProfile:
"""Fetches the user by its visible-to-other users username (in the
`email` field). For use in API contexts; do not use in
authentication/registration contexts as doing so will break
authentication in organizations using
EMAIL_ADDRESS_VISIBILITY_ADMINS. In those code paths, use
get_user_by_delivery_email.
"""
return UserProfile.objects.select_related(
"realm",
"realm__can_access_all_users_group",
"realm__can_access_all_users_group__named_user_group",
"realm__direct_message_initiator_group",
"realm__direct_message_initiator_group__named_user_group",
"realm__direct_message_permission_group",
"realm__direct_message_permission_group__named_user_group",
"bot_owner",
).get(email__iexact=email.strip(), realm=realm)
def get_active_user(email: str, realm: "Realm") -> UserProfile:
"""Variant of get_user_by_email that excludes deactivated users.
See get_user docstring for important usage notes."""
user_profile = get_user(email, realm)
if not user_profile.is_active:
raise UserProfile.DoesNotExist
return user_profile
def get_user_profile_by_id_in_realm(uid: int, realm: "Realm") -> UserProfile:
return UserProfile.objects.select_related(
"realm",
"realm__can_access_all_users_group",
"realm__can_access_all_users_group__named_user_group",
"bot_owner",
).get(id=uid, realm=realm)
def get_active_user_profile_by_id_in_realm(uid: int, realm: "Realm") -> UserProfile:
user_profile = get_user_profile_by_id_in_realm(uid, realm)
if not user_profile.is_active:
raise UserProfile.DoesNotExist
return user_profile
def get_user_including_cross_realm(email: str, realm: "Realm") -> UserProfile:
if is_cross_realm_bot_email(email):
return get_system_bot(email, realm.id)
assert realm is not None
return get_user(email, realm)
@cache_with_key(bot_profile_cache_key, timeout=3600 * 24 * 7)
def get_system_bot(email: str, realm_id: int) -> UserProfile:
"""
This function doesn't use the realm_id argument yet, but requires
passing it as preparation for adding system bots to each realm instead
of having them all in a separate system bot realm.
If you're calling this function, use the id of the realm in which the system
bot will be after that migration. If the bot is supposed to send a message,
the same realm as the one *to* which the message will be sent should be used - because
cross-realm messages will be eliminated as part of the migration.
"""
return UserProfile.objects.select_related("realm").get(email__iexact=email.strip())
def get_user_by_id_in_realm_including_cross_realm(
uid: int,
realm: Optional["Realm"],
) -> UserProfile:
user_profile = get_user_profile_by_id(uid)
if user_profile.realm == realm:
return user_profile
# Note: This doesn't validate whether the `realm` passed in is
# None/invalid for the is_cross_realm_bot_email case.
if is_cross_realm_bot_email(user_profile.delivery_email):
return user_profile
raise UserProfile.DoesNotExist
@cache_with_key(realm_user_dicts_cache_key, timeout=3600 * 24 * 7)
def get_realm_user_dicts(realm_id: int) -> list[RawUserDict]:
return list(
UserProfile.objects.filter(
realm_id=realm_id,
).values(*realm_user_dict_fields)
)
@cache_with_key(active_user_ids_cache_key, timeout=3600 * 24 * 7)
def active_user_ids(realm_id: int) -> list[int]:
query = UserProfile.objects.filter(
realm_id=realm_id,
is_active=True,
).values_list("id", flat=True)
return list(query)
@cache_with_key(active_non_guest_user_ids_cache_key, timeout=3600 * 24 * 7)
def active_non_guest_user_ids(realm_id: int) -> list[int]:
query = (
UserProfile.objects.filter(
realm_id=realm_id,
is_active=True,
)
.exclude(
role=UserProfile.ROLE_GUEST,
)
.values_list("id", flat=True)
)
return list(query)
def bot_owner_user_ids(user_profile: UserProfile) -> set[int]:
is_private_bot = (
user_profile.default_sending_stream and user_profile.default_sending_stream.invite_only
) or (
user_profile.default_events_register_stream
and user_profile.default_events_register_stream.invite_only
)
if is_private_bot:
if user_profile.bot_owner_id is not None:
return {user_profile.bot_owner_id}
return set()
else:
users = {user.id for user in user_profile.realm.get_human_admin_users()}
if user_profile.bot_owner_id is not None:
users.add(user_profile.bot_owner_id)
return users
def get_source_profile(email: str, realm_id: int) -> UserProfile | None:
from zerver.models import Realm
from zerver.models.realms import get_realm_by_id
try:
return get_user_by_delivery_email(email, get_realm_by_id(realm_id))
except (Realm.DoesNotExist, UserProfile.DoesNotExist):
return None
@cache_with_key(lambda realm: bot_dicts_in_realm_cache_key(realm.id), timeout=3600 * 24 * 7)
def get_bot_dicts_in_realm(realm: "Realm") -> list[dict[str, Any]]:
return list(UserProfile.objects.filter(realm=realm, is_bot=True).values(*bot_dict_fields))
def is_cross_realm_bot_email(email: str) -> bool:
return email.lower() in settings.CROSS_REALM_BOT_EMAILS
Reference in New Issue View Git Blame Copy Permalink