import base64 import hmac import secrets import arrow from flask import render_template, redirect, url_for, flash, request from flask_login import login_required, current_user from app import config from app.dashboard.base import dashboard_bp from app.dashboard.views.enter_sudo import sudo_required from app.db import Session from app.email_utils import send_email, render from app.extensions import limiter from app.log import LOG from app.models import ( Alias, Contact, AliasUsedOn, AliasMailbox, User, ClientUser, ) from app.models import Mailbox from app.utils import CSRFValidationForm def transfer(alias, new_user, new_mailboxes: [Mailbox]): # cannot transfer alias which is used for receiving newsletter if User.get_by(newsletter_alias_id=alias.id): raise Exception("Cannot transfer alias that's used to receive newsletter") # update user_id Session.query(Contact).filter(Contact.alias_id == alias.id).update( {"user_id": new_user.id} ) Session.query(AliasUsedOn).filter(AliasUsedOn.alias_id == alias.id).update( {"user_id": new_user.id} ) Session.query(ClientUser).filter(ClientUser.alias_id == alias.id).update( {"user_id": new_user.id} ) # remove existing mailboxes from the alias Session.query(AliasMailbox).filter(AliasMailbox.alias_id == alias.id).delete() # set mailboxes alias.mailbox_id = new_mailboxes.pop().id for mb in new_mailboxes: AliasMailbox.create(alias_id=alias.id, mailbox_id=mb.id) # alias has never been transferred before if not alias.original_owner_id: alias.original_owner_id = alias.user_id # inform previous owner old_user = alias.user send_email( old_user.email, f"Alias {alias.email} has been received", render( "transactional/alias-transferred.txt", alias=alias, ), render( "transactional/alias-transferred.html", alias=alias, ), ) # now the alias belongs to the new user alias.user_id = new_user.id # set some fields back to default alias.disable_pgp = False alias.pinned = False Session.commit() def hmac_alias_transfer_token(transfer_token: str) -> str: alias_hmac = hmac.new( config.ALIAS_TRANSFER_TOKEN_SECRET.encode("utf-8"), transfer_token.encode("utf-8"), "sha3_224", ) return base64.urlsafe_b64encode(alias_hmac.digest()).decode("utf-8").rstrip("=") @dashboard_bp.route("/alias_transfer/send//", methods=["GET", "POST"]) @login_required @sudo_required def alias_transfer_send_route(alias_id): alias = Alias.get(alias_id) if not alias or alias.user_id != current_user.id: flash("You cannot see this page", "warning") return redirect(url_for("dashboard.index")) if current_user.newsletter_alias_id == alias.id: flash( "This alias is currently used for receiving the newsletter and cannot be transferred", "error", ) return redirect(url_for("dashboard.index")) alias_transfer_url = None csrf_form = CSRFValidationForm() if request.method == "POST": if not csrf_form.validate(): flash("Invalid request", "warning") return redirect(request.url) # generate a new transfer_token if request.form.get("form-name") == "create": transfer_token = f"{alias.id}.{secrets.token_urlsafe(32)}" alias.transfer_token = hmac_alias_transfer_token(transfer_token) alias.transfer_token_expiration = arrow.utcnow().shift(hours=24) Session.commit() alias_transfer_url = ( config.URL + "/dashboard/alias_transfer/receive" + f"?token={transfer_token}" ) flash("Share alias URL created", "success") # request.form.get("form-name") == "remove" else: alias.transfer_token = None alias.transfer_token_expiration = None Session.commit() alias_transfer_url = None flash("Share URL deleted", "success") return render_template( "dashboard/alias_transfer_send.html", alias=alias, alias_transfer_url=alias_transfer_url, link_active=alias.transfer_token_expiration is not None and alias.transfer_token_expiration > arrow.utcnow(), csrf_form=csrf_form, ) @dashboard_bp.route("/alias_transfer/receive", methods=["GET", "POST"]) @limiter.limit("5/minute") @login_required def alias_transfer_receive_route(): """ URL has ?alias_id=signed_alias_id """ token = request.args.get("token") if not token: flash("Invalid transfer token", "error") return redirect(url_for("dashboard.index")) hashed_token = hmac_alias_transfer_token(token) # TODO: Don't allow unhashed tokens once all the tokens have been migrated to the new format alias = Alias.get_by(transfer_token=token) or Alias.get_by( transfer_token=hashed_token ) if not alias: flash("Invalid link", "error") return redirect(url_for("dashboard.index")) # TODO: Don't allow none once all the tokens have been migrated to the new format if ( alias.transfer_token_expiration is not None and alias.transfer_token_expiration < arrow.utcnow() ): flash("Expired link, please request a new one", "error") return redirect(url_for("dashboard.index")) # alias already belongs to this user if alias.user_id == current_user.id: flash("You already own this alias", "warning") return redirect(url_for("dashboard.index")) # check if user has not exceeded the alias quota if not current_user.can_create_new_alias(): LOG.d("%s can't receive new alias", current_user) flash( "You have reached free plan limit, please upgrade to create new aliases", "warning", ) return redirect(url_for("dashboard.index")) mailboxes = current_user.mailboxes() if request.method == "POST": mailbox_ids = request.form.getlist("mailbox_ids") # check if mailbox is not tempered with mailboxes = [] for mailbox_id in mailbox_ids: mailbox = Mailbox.get(mailbox_id) if ( not mailbox or mailbox.user_id != current_user.id or not mailbox.verified ): flash("Something went wrong, please retry", "warning") return redirect(request.url) mailboxes.append(mailbox) if not mailboxes: flash("You must select at least 1 mailbox", "warning") return redirect(request.url) LOG.d( "transfer alias %s from %s to %s with %s with token %s", alias, alias.user, current_user, mailboxes, token, ) transfer(alias, current_user, mailboxes) flash(f"You are now owner of {alias.email}", "success") return redirect(url_for("dashboard.index", highlight_alias_id=alias.id)) return render_template( "dashboard/alias_transfer_receive.html", alias=alias, mailboxes=mailboxes, )