Code reference › domain › person

module

pretalx.person.domain.verification

Functions 14

cancel_email_change(user)source

def cancel_email_change(user):
    if not user.pending_email:
        return
    pending = user.pending_email
    _clear_pending_email(user)
    user.log_action("pretalx.user.email.change.cancel", data={"pending_email": pending})

check_send_cooldown(user, *, free_pass=False)source

def check_send_cooldown(user, *, free_pass=False):
    # One correction is free, afterwards follow resend cooldown timer
    remaining = send_cooldown_remaining(user)
    if not remaining:
        return
    if free_pass and cache.add(
        _free_correction_key(user), True, timeout=dt.timedelta(hours=24).total_seconds()
    ):
        return
    raise SendCooldownError(remaining)

confirm_verification(token)source

def confirm_verification(token):
    user, kind = parse_verification_token(token)

    if kind == KIND_CHANGE:
        if pending_email_expired(user):
            _clear_pending_email(user)
            raise PendingEmailExpiredError
        address = user.pending_email
        try:
            validate_email_unique(address, exclude_user=user)
        except ValidationError as error:
            _clear_pending_email(user)
            user.log_action(
                "pretalx.user.email.change.taken", data={"pending_email": address}
            )
            raise PendingEmailTakenError from error
        with transaction.atomic():
            old_email = user.email
            user.email = address
            user.pending_email = None
            user.pending_email_sent = None
            user.email_verification_state = EmailVerificationState.VERIFIED
            user.save(
                update_fields=[
                    "email",
                    "pending_email",
                    "pending_email_sent",
                    "email_verification_state",
                ]
            )
            send_system_mail(
                subject=EMAIL_CHANGED_SUBJECT,
                text=EMAIL_CHANGED_TEXT,
                to=old_email,
                locale=user.locale,
                # Django's EmailField accepts RFC 5321 quoted local parts, which
                # could include e.g. "<script>", so we treat them as untrusted
                # and escape them with prejudice.
                safe_extra_context={
                    "old_email": untrusted_plain_value(old_email),
                    "new_email": untrusted_plain_value(user.email),
                },
                context_kwargs={"user": user},
            )
            user.log_action(
                "pretalx.user.email.change.confirm",
                data={"old_email": old_email, "new_email": user.email},
            )
    else:
        if user.email_verification_state == EmailVerificationState.VERIFIED:
            raise AlreadyVerifiedError
        with transaction.atomic():
            user.email_verification_state = EmailVerificationState.VERIFIED
            user.save(update_fields=["email_verification_state"])
            user.log_action(
                "pretalx.user.email.verification.confirm", data={"email": user.email}
            )
    return user, kind

correct_unverified_email(user, address, *, event=None, orga=False)source

@transaction.atomic
def correct_unverified_email(user, address, *, event=None, orga=False):
    if user.email_verification_state != EmailVerificationState.UNVERIFIED:
        raise VerificationError(
            "Only unverified accounts may correct their address directly"
        )
    address = address.lower().strip()
    try:
        validate_email_unique(address, exclude_user=user)
    except ValidationError as error:
        raise PendingEmailTakenError from error
    old_email = user.email
    user.email = address
    user.save(update_fields=["email"])
    send_verification_mail(user, KIND_VERIFY, event=event, orga=orga, free_pass=True)
    user.log_action(
        "pretalx.user.email.verification.correct",
        data={"old_email": old_email, "new_email": user.email},
    )

finalize_registration(user, *, event=None, orga=False, invited_email=None)source

def finalize_registration(user, *, event=None, orga=False, invited_email=None):
    if promote_on_invitation_match(user, invited_email):
        return
    with suppress(SendCooldownError):
        send_verification_mail(user, KIND_VERIFY, event=event, orga=orga)

get_verification_url(token, *, event=None, orga=False)source

def get_verification_url(token, *, event=None, orga=False):
    if event:
        path = "orga:event.auth.verify" if orga else "cfp:event.verify"
        kwargs = {"token": token, "event": event.slug}
    else:
        path = "orga:auth.verify"
        kwargs = {"token": token}
    return UrlString(build_absolute_uri(path, kwargs=kwargs))

make_verification_token(user, kind)source

def make_verification_token(user, kind):
    if kind == KIND_CHANGE and not user.pending_email:
        raise ValueError("Cannot make a change token without a pending email address")
    return signing.dumps(
        {"user": user.code, "kind": kind, "check": _token_check(user, kind)},
        salt=VERIFICATION_SALT,
    )

parse_verification_token(token)source

def parse_verification_token(token):
    try:
        payload = signing.loads(
            token, salt=VERIFICATION_SALT, max_age=dt.timedelta(hours=24)
        )
    except signing.SignatureExpired as error:
        raise ExpiredVerificationTokenError from error
    except signing.BadSignature as error:
        raise InvalidVerificationTokenError from error
    user = User.objects.filter(code=payload.get("user")).first()
    kind = payload.get("kind")
    if not user or kind not in (KIND_VERIFY, KIND_CHANGE):
        raise InvalidVerificationTokenError
    if not constant_time_compare(payload.get("check", ""), _token_check(user, kind)):
        raise InvalidVerificationTokenError
    return user, kind

pending_email_expired(user)source

def pending_email_expired(user):
    return bool(user.pending_email_sent) and (
        now() - user.pending_email_sent > dt.timedelta(hours=24)
    )

promote_on_invitation_match(user, invited_email)source

def promote_on_invitation_match(user, invited_email):
    if not invited_email:
        return False
    if invited_email.strip().lower() != user.email.lower():
        return False
    if user.email_verification_state != EmailVerificationState.VERIFIED:
        user.email_verification_state = EmailVerificationState.VERIFIED
        user.save(update_fields=["email_verification_state"])
        user.log_action(
            "pretalx.user.email.verification.promote", data={"email": user.email}
        )
    return True

request_email_change(user, address, *, event=None, orga=False)source

@transaction.atomic
def request_email_change(user, address, *, event=None, orga=False):
    address = address.lower().strip()
    validate_email_unique(address, exclude_user=user)
    user.pending_email = address
    user.pending_email_sent = now()
    user.save(update_fields=["pending_email", "pending_email_sent"])
    send_verification_mail(user, KIND_CHANGE, event=event, orga=orga)
    send_system_mail(
        subject=EMAIL_CHANGE_REQUESTED_SUBJECT,
        text=EMAIL_CHANGE_REQUESTED_TEXT,
        to=user.email,
        event=event,
        locale=user.locale,
        safe_extra_context={"pending_email": untrusted_plain_value(address)},
        context_kwargs={"user": user},
    )
    user.log_action(
        "pretalx.user.email.change.request", data={"pending_email": address}
    )

send_cooldown_remaining(user)source

def send_cooldown_remaining(user):
    sent = cache.get(_cooldown_key(user))
    if not sent:
        return 0
    remaining = SEND_COOLDOWN.total_seconds() - (now().timestamp() - sent)
    return max(0, math.ceil(remaining))

send_verification_mail(user, kind, *, event=None, orga=False, free_pass=False)source

def send_verification_mail(user, kind, *, event=None, orga=False, free_pass=False):
    to = user.pending_email if kind == KIND_CHANGE else user.email
    if not to:
        raise ValueError("Cannot send a change link without a pending email address")
    check_send_cooldown(user, free_pass=free_pass)
    token = make_verification_token(user, kind)
    send_system_mail(
        subject=EMAIL_VERIFICATION_SUBJECT,
        text=EMAIL_VERIFICATION_TEXT,
        to=to,
        event=event,
        locale=user.locale,
        safe_extra_context={"url": get_verification_url(token, event=event, orga=orga)},
        context_kwargs={"user": user},
    )
    cache.set(
        _cooldown_key(user), now().timestamp(), timeout=SEND_COOLDOWN.total_seconds()
    )
    user.log_action(
        "pretalx.user.email.verification.send", data={"email": to, "kind": kind}
    )

set_verification_state(user, state, *, actor)source

def set_verification_state(user, state, *, actor):
    if state not in (
        EmailVerificationState.VERIFIED,
        EmailVerificationState.UNVERIFIED,
    ):
        raise ValueError(
            "Administrators may only set accounts to verified or unverified"
        )
    if user.email_verification_state == state:
        return
    user.email_verification_state = state
    user.save(update_fields=["email_verification_state"])
    user.log_action(
        "pretalx.user.email.verification.set",
        person=actor,
        orga=True,
        data={"state": state},
    )