237 lines
6.8 KiB
Python
237 lines
6.8 KiB
Python
from smtplib import SMTPRecipientsRefused
|
|
|
|
import arrow
|
|
from flask import g
|
|
from flask import jsonify
|
|
from flask import request
|
|
|
|
from app.api.base import api_bp, require_api_auth
|
|
from app.config import JOB_DELETE_MAILBOX
|
|
from app.dashboard.views.mailbox import send_verification_email
|
|
from app.dashboard.views.mailbox_detail import verify_mailbox_change
|
|
from app.db import Session
|
|
from app.email_utils import (
|
|
mailbox_already_used,
|
|
email_can_be_used_as_mailbox,
|
|
is_valid_email,
|
|
)
|
|
from app.log import LOG
|
|
from app.models import Mailbox, Job
|
|
from app.utils import sanitize_email
|
|
|
|
|
|
def mailbox_to_dict(mailbox: Mailbox):
|
|
return {
|
|
"id": mailbox.id,
|
|
"email": mailbox.email,
|
|
"verified": mailbox.verified,
|
|
"default": mailbox.user.default_mailbox_id == mailbox.id,
|
|
"creation_timestamp": mailbox.created_at.timestamp,
|
|
"nb_alias": mailbox.nb_alias(),
|
|
}
|
|
|
|
|
|
@api_bp.route("/mailboxes", methods=["POST"])
|
|
@require_api_auth
|
|
def create_mailbox():
|
|
"""
|
|
Create a new mailbox. User needs to verify the mailbox via an activation email.
|
|
Input:
|
|
email: in body
|
|
Output:
|
|
the new mailbox dict
|
|
"""
|
|
user = g.user
|
|
mailbox_email = sanitize_email(request.get_json().get("email"))
|
|
|
|
if not user.is_premium():
|
|
return jsonify(error=f"Only premium plan can add additional mailbox"), 400
|
|
|
|
if not is_valid_email(mailbox_email):
|
|
return jsonify(error=f"{mailbox_email} invalid"), 400
|
|
elif mailbox_already_used(mailbox_email, user):
|
|
return jsonify(error=f"{mailbox_email} already used"), 400
|
|
elif not email_can_be_used_as_mailbox(mailbox_email):
|
|
return (
|
|
jsonify(
|
|
error=f"{mailbox_email} cannot be used. Please note a mailbox cannot "
|
|
f"be a disposable email address"
|
|
),
|
|
400,
|
|
)
|
|
else:
|
|
new_mailbox = Mailbox.create(email=mailbox_email, user_id=user.id)
|
|
Session.commit()
|
|
|
|
send_verification_email(user, new_mailbox)
|
|
|
|
return (
|
|
jsonify(mailbox_to_dict(new_mailbox)),
|
|
201,
|
|
)
|
|
|
|
|
|
@api_bp.route("/mailboxes/<int:mailbox_id>", methods=["DELETE"])
|
|
@require_api_auth
|
|
def delete_mailbox(mailbox_id):
|
|
"""
|
|
Delete mailbox
|
|
Input:
|
|
mailbox_id: in url
|
|
(optional) transfer_aliases_to: in body. Id of the new mailbox for the aliases.
|
|
If omitted or the value is set to -1,
|
|
the aliases of the mailbox will be deleted too.
|
|
Output:
|
|
200 if deleted successfully
|
|
|
|
"""
|
|
user = g.user
|
|
mailbox = Mailbox.get(id=mailbox_id)
|
|
|
|
if not mailbox or mailbox.user_id != user.id:
|
|
return jsonify(error="Forbidden"), 403
|
|
|
|
if mailbox.id == user.default_mailbox_id:
|
|
return jsonify(error="You cannot delete the default mailbox"), 400
|
|
|
|
data = request.get_json() or {}
|
|
transfer_mailbox_id = data.get("transfer_aliases_to")
|
|
if transfer_mailbox_id and int(transfer_mailbox_id) >= 0:
|
|
transfer_mailbox = Mailbox.get(transfer_mailbox_id)
|
|
|
|
if not transfer_mailbox or transfer_mailbox.user_id != user.id:
|
|
return (
|
|
jsonify(error="You must transfer the aliases to a mailbox you own."),
|
|
403,
|
|
)
|
|
|
|
if transfer_mailbox_id == mailbox_id:
|
|
return (
|
|
jsonify(
|
|
error="You can not transfer the aliases to the mailbox you want to delete."
|
|
),
|
|
400,
|
|
)
|
|
|
|
if not transfer_mailbox.verified:
|
|
return jsonify(error="Your new mailbox is not verified"), 400
|
|
|
|
# Schedule delete account job
|
|
LOG.w("schedule delete mailbox job for %s", mailbox)
|
|
Job.create(
|
|
name=JOB_DELETE_MAILBOX,
|
|
payload={
|
|
"mailbox_id": mailbox.id,
|
|
"transfer_mailbox_id": transfer_mailbox_id,
|
|
},
|
|
run_at=arrow.now(),
|
|
commit=True,
|
|
)
|
|
|
|
return jsonify(deleted=True), 200
|
|
|
|
|
|
@api_bp.route("/mailboxes/<int:mailbox_id>", methods=["PUT"])
|
|
@require_api_auth
|
|
def update_mailbox(mailbox_id):
|
|
"""
|
|
Update mailbox
|
|
Input:
|
|
mailbox_id: in url
|
|
(optional) default: in body. Set a mailbox as the default mailbox.
|
|
(optional) email: in body. Change a mailbox email.
|
|
(optional) cancel_email_change: in body. Cancel mailbox email change.
|
|
Output:
|
|
200 if updated successfully
|
|
|
|
"""
|
|
user = g.user
|
|
mailbox = Mailbox.get(mailbox_id)
|
|
|
|
if not mailbox or mailbox.user_id != user.id:
|
|
return jsonify(error="Forbidden"), 403
|
|
|
|
data = request.get_json() or {}
|
|
changed = False
|
|
if "default" in data:
|
|
is_default = data.get("default")
|
|
if is_default:
|
|
if not mailbox.verified:
|
|
return (
|
|
jsonify(
|
|
error="Unverified mailbox cannot be used as default mailbox"
|
|
),
|
|
400,
|
|
)
|
|
user.default_mailbox_id = mailbox.id
|
|
changed = True
|
|
|
|
if "email" in data:
|
|
new_email = sanitize_email(data.get("email"))
|
|
|
|
if mailbox_already_used(new_email, user):
|
|
return jsonify(error=f"{new_email} already used"), 400
|
|
elif not email_can_be_used_as_mailbox(new_email):
|
|
return (
|
|
jsonify(
|
|
error=f"{new_email} cannot be used. Please note a mailbox cannot "
|
|
f"be a disposable email address"
|
|
),
|
|
400,
|
|
)
|
|
|
|
try:
|
|
verify_mailbox_change(user, mailbox, new_email)
|
|
except SMTPRecipientsRefused:
|
|
return jsonify(error=f"Incorrect mailbox, please recheck {new_email}"), 400
|
|
else:
|
|
mailbox.new_email = new_email
|
|
changed = True
|
|
|
|
if "cancel_email_change" in data:
|
|
cancel_email_change = data.get("cancel_email_change")
|
|
if cancel_email_change:
|
|
mailbox.new_email = None
|
|
changed = True
|
|
|
|
if changed:
|
|
Session.commit()
|
|
|
|
return jsonify(updated=True), 200
|
|
|
|
|
|
@api_bp.route("/mailboxes", methods=["GET"])
|
|
@require_api_auth
|
|
def get_mailboxes():
|
|
"""
|
|
Get verified mailboxes
|
|
Output:
|
|
- mailboxes: list of mailbox dict
|
|
"""
|
|
user = g.user
|
|
|
|
return (
|
|
jsonify(mailboxes=[mailbox_to_dict(mb) for mb in user.mailboxes()]),
|
|
200,
|
|
)
|
|
|
|
|
|
@api_bp.route("/v2/mailboxes", methods=["GET"])
|
|
@require_api_auth
|
|
def get_mailboxes_v2():
|
|
"""
|
|
Get all mailboxes - including unverified mailboxes
|
|
Output:
|
|
- mailboxes: list of mailbox dict
|
|
"""
|
|
user = g.user
|
|
mailboxes = []
|
|
|
|
for mailbox in Mailbox.filter_by(user_id=user.id):
|
|
mailboxes.append(mailbox)
|
|
|
|
return (
|
|
jsonify(mailboxes=[mailbox_to_dict(mb) for mb in mailboxes]),
|
|
200,
|
|
)
|