blender-id/bid_api/signals.py

import functools
import logging


from django.contrib.auth import get_user_model
from django.core import serializers
from django.db.models.signals import pre_save, post_save
from django.dispatch import receiver, Signal

from . import models
import bid_api.tasks
import bid_main.file_utils

log = logging.getLogger(__name__)
UserModel = get_user_model()
# This signal provides the following arguments: ["user", "old_email"]
user_email_changed = Signal()

USER_SAVE_INTERESTING_FIELDS = {
    "email",
    "full_name",
    "public_roles_as_string",
    "avatar",
    "date_deletion_requested",
    "confirmed_email_at",
    "nickname",
}


def _get_object_state(obj: object, fields=None, include_pk=False) -> dict:
    data = serializers.serialize('python', [obj], fields=fields)[0]
    if include_pk:
        data['fields']['pk'] = data['pk']
    return data['fields']


def filter_user_save_hook(wrapped):
    """Decorator for the webhook user-save signal handlers."""

    @functools.wraps(wrapped)
    def wrapper(sender, instance, *args, **kwargs):
        if sender != UserModel or not isinstance(instance, UserModel):
            # log.debug('skipping save of sender %r', sender)
            return

        update_fields = kwargs.get("update_fields")
        if update_fields is not None:
            # log.debug('User %s was saved on fields %r only', instance.email, update_fields)
            if not set(update_fields).intersection(USER_SAVE_INTERESTING_FIELDS):
                # log.debug('User %s was modified only on ignored fields; ignoring', instance.email)
                return

        return wrapped(sender, instance, *args, **kwargs)

    return wrapper


def _check_user_modification(new_user: UserModel, old_user: UserModel, **kwargs) -> bool:
    """Returns True iff the user was modified.

    Only checks fields listed in USER_SAVE_INTERESTING_FIELDS.
    """
    old_state = _get_object_state(old_user, fields=USER_SAVE_INTERESTING_FIELDS)
    new_state = _get_object_state(new_user, fields=USER_SAVE_INTERESTING_FIELDS)
    update_fields = kwargs.get('update_fields')
    changed_fields = set()
    for field in USER_SAVE_INTERESTING_FIELDS:
        if update_fields is not None and field not in update_fields:
            continue
        if new_state.get(field) != old_state.get(field):
            changed_fields.add(field)
    log.info('%s are being changed on pk=%s', changed_fields, new_user.pk)
    return len(changed_fields) > 0, old_state


@receiver(pre_save)
@filter_user_save_hook
def inspect_modified_user(sender, user: UserModel, **kwargs):
    # update_fields = kwargs.get('update_fields')
    # log.debug('pre-save of %s, fields %s', user.email, update_fields)

    # Default to False
    user.webhook_user_modified = False

    if not user.id:
        # New user; don't notify the webhooks.
        # log.debug('%s is a new user (no ID yet), skipping', user.email)
        return
    try:
        db_user = UserModel.objects.get(id=user.id)
    except UserModel.DoesNotExist:
        # New user; don't notify the webhooks.
        # log.debug('%s is a new user (not in DB), skipping', user.email)
        return

    # Make sure that the post-save hook knows what the pre-save user looks like.
    was_modified, old_state = _check_user_modification(user, db_user, **kwargs)
    user.webhook_user_modified = was_modified
    user.webhook_pre_save = old_state


@receiver(post_save)
@filter_user_save_hook
def handle_modifications(sender, user: UserModel, **kwargs):
    """Act on various account modifications.

    Clean up orphaned avatars, notify about email change and call webhooks.
    The payload is POSTed, and a HMAC-SHA256 checksum is sent using the
    X-Webhook-HMAC HTTP header.

    Also see https://docs.djangoproject.com/en/1.11/ref/signals/#post-save
    """

    # update_fields = kwargs.get('update_fields')
    # log.debug('post-save of %s, fields %s', user.email, update_fields)

    if not getattr(user, "webhook_user_modified", False):
        # log.debug('Skipping save of %s', user.email)
        return

    # Get the old email address so that the webhook receiver can match by
    # either database ID or email address.
    webhook_pre_save = getattr(user, "webhook_pre_save", {})
    old_email = webhook_pre_save.get("email")
    old_nickname = webhook_pre_save.get("nickname") or ""

    # Map all falsey values to empty string for more consistent comparison later.
    # An empty avatar can be either '' or None.
    old_avatar = webhook_pre_save.get("avatar") or ""
    cur_avatar = user.avatar or ""

    if old_avatar and old_avatar != cur_avatar:
        # Only delete the avatar if a new one has been set. The user can be
        # saved without fetching the avatar from the database, and in such a
        # case it should not be deleted.
        log.debug(
            "User changed avatar to %r, going to delete old avatar file %r",
            cur_avatar,
            old_avatar,
        )
        bid_main.file_utils.delete_avatar_files(old_avatar)

    if old_email != user.email:
        log.debug("User changed email from %s to %s", old_email, user.email)
        user_email_changed.send(sender, user=user, old_email=old_email)

    hooks = list(
        models.Webhook.objects.filter(enabled=True).filter(
            app__in=user.bid_main_oauth2accesstoken.distinct().values_list('application', flat=True)
        )
    )

    hooks_count = len(hooks)
    if hooks_count == 0:
        log.info('Not sending modification of pk=%d to any webhooks', user.pk)
        return

    log.info("Sending modification of pk=%d to %d webhooks", user.pk, hooks_count)

    # Do our own JSON encoding so that we can compute the HMAC using the hook's secret.
    payload = {
        "id": user.id,
        "old_email": old_email,
        "full_name": user.get_full_name(),
        "email": user.email,
        "roles": sorted(user.public_roles()),
        "avatar_changed": old_avatar != cur_avatar,
        "avatar_url": user.avatar_thumbnail_url() if user.avatar else None,
        "date_deletion_requested": isoformat(user.date_deletion_requested),
        "confirmed_email_at": isoformat(user.confirmed_email_at),
        "old_nickname": old_nickname,
        "nickname": user.nickname,
    }

    for hook in hooks:
        log.debug("Will send to %s, %s", hook, hook.url)
        verbose_name = f'Webhook pk={hook.pk} "{hook}"'
        bid_api.tasks.webhook_send(hook.pk, payload, verbose_name=verbose_name, creator=hook)


def isoformat(timestamp):
    """None-safe timestamp.isoformat()."""
    if not timestamp:
        return None
    return timestamp.isoformat()