app-MAIL-temp/app/api/views/alias.py

387 lines
9.2 KiB
Python
Raw Normal View History

from email.utils import parseaddr
2020-02-04 17:26:59 +01:00
from flask import g
from flask import jsonify
from flask import request
2020-02-04 17:26:59 +01:00
from flask_cors import cross_origin
from app.api.base import api_bp, verify_api_key
from app.config import EMAIL_DOMAIN
2020-03-14 12:22:43 +01:00
from app.config import PAGE_LIMIT
from app.dashboard.views.alias_log import get_alias_log
2020-03-17 20:16:20 +01:00
from app.dashboard.views.index import (
AliasInfo,
get_alias_infos_with_pagination,
2020-03-26 19:44:00 +01:00
get_alias_info,
2020-03-17 20:16:20 +01:00
)
2020-02-04 17:26:59 +01:00
from app.extensions import db
2020-03-14 12:55:38 +01:00
from app.log import LOG
from app.models import Alias, Contact
2020-03-17 20:16:20 +01:00
from app.models import EmailLog
2020-03-14 12:55:38 +01:00
from app.utils import random_string
2020-02-04 17:26:59 +01:00
@api_bp.route("/aliases", methods=["GET", "POST"])
2020-02-04 17:26:59 +01:00
@cross_origin()
@verify_api_key
def get_aliases():
"""
Get aliases
Input:
page_id: in query
Output:
- aliases: list of alias:
- id
- email
- creation_date
- creation_timestamp
- nb_forward
- nb_block
- nb_reply
2020-03-11 12:13:38 +01:00
- note
2020-02-04 17:26:59 +01:00
"""
user = g.user
try:
page_id = int(request.args.get("page_id"))
except (ValueError, TypeError):
return jsonify(error="page_id must be provided in request query"), 400
query = None
data = request.get_json(silent=True)
if data:
query = data.get("query")
2020-03-17 20:16:20 +01:00
alias_infos: [AliasInfo] = get_alias_infos_with_pagination(
user, page_id=page_id, query=query
)
2020-02-04 17:26:59 +01:00
return (
jsonify(
aliases=[serialize_alias_info(alias_info) for alias_info in alias_infos]
2020-02-04 17:26:59 +01:00
),
200,
)
def serialize_alias_info(alias_info: AliasInfo) -> dict:
return {
"id": alias_info.id,
"email": alias_info.alias.email,
"creation_date": alias_info.alias.created_at.format(),
"creation_timestamp": alias_info.alias.created_at.timestamp,
"nb_forward": alias_info.nb_forward,
"nb_block": alias_info.nb_blocked,
"nb_reply": alias_info.nb_reply,
"enabled": alias_info.alias.enabled,
"note": alias_info.note,
}
@api_bp.route("/aliases/<int:alias_id>", methods=["DELETE"])
@cross_origin()
@verify_api_key
def delete_alias(alias_id):
"""
Delete alias
Input:
alias_id: in url
Output:
200 if deleted successfully
"""
user = g.user
alias = Alias.get(alias_id)
if alias.user_id != user.id:
return jsonify(error="Forbidden"), 403
Alias.delete(alias_id)
db.session.commit()
return jsonify(deleted=True), 200
@api_bp.route("/aliases/<int:alias_id>/toggle", methods=["POST"])
@cross_origin()
@verify_api_key
def toggle_alias(alias_id):
"""
Enable/disable alias
Input:
alias_id: in url
Output:
200 along with new status:
- enabled
"""
user = g.user
alias: Alias = Alias.get(alias_id)
if alias.user_id != user.id:
return jsonify(error="Forbidden"), 403
alias.enabled = not alias.enabled
db.session.commit()
return jsonify(enabled=alias.enabled), 200
@api_bp.route("/aliases/<int:alias_id>/activities")
@cross_origin()
@verify_api_key
def get_alias_activities(alias_id):
"""
Get aliases
Input:
page_id: in query
Output:
- activities: list of activity:
- from
- to
- timestamp
- action: forward|reply|block
- reverse_alias
"""
user = g.user
try:
page_id = int(request.args.get("page_id"))
except (ValueError, TypeError):
return jsonify(error="page_id must be provided in request query"), 400
alias: Alias = Alias.get(alias_id)
if alias.user_id != user.id:
return jsonify(error="Forbidden"), 403
alias_logs = get_alias_log(alias, page_id)
activities = []
for alias_log in alias_logs:
activity = {
"timestamp": alias_log.when.timestamp,
"reverse_alias": alias_log.reverse_alias,
}
if alias_log.is_reply:
activity["from"] = alias_log.alias
activity["to"] = alias_log.website_from or alias_log.website_email
activity["action"] = "reply"
else:
activity["to"] = alias_log.alias
activity["from"] = alias_log.website_from or alias_log.website_email
2020-02-22 07:27:22 +01:00
if alias_log.bounced:
activity["action"] = "bounced"
elif alias_log.blocked:
activity["action"] = "block"
else:
activity["action"] = "forward"
activities.append(activity)
2020-03-14 12:22:43 +01:00
return jsonify(activities=activities), 200
2020-03-14 11:38:39 +01:00
@api_bp.route("/aliases/<int:alias_id>", methods=["PUT"])
@cross_origin()
@verify_api_key
def update_alias(alias_id):
"""
Update alias note
Input:
alias_id: in url
note: in body
Output:
200
"""
data = request.get_json()
if not data:
return jsonify(error="request body cannot be empty"), 400
user = g.user
alias: Alias = Alias.get(alias_id)
2020-03-14 11:38:39 +01:00
if alias.user_id != user.id:
2020-03-14 11:38:39 +01:00
return jsonify(error="Forbidden"), 403
new_note = data.get("note")
alias.note = new_note
2020-03-14 11:38:39 +01:00
db.session.commit()
return jsonify(note=new_note), 200
2020-03-14 12:22:43 +01:00
2020-03-26 19:44:00 +01:00
@api_bp.route("/aliases/<int:alias_id>", methods=["GET"])
@cross_origin()
@verify_api_key
def get_alias(alias_id):
"""
Get alias
Input:
alias_id: in url
Output:
Alias info, same as in get_aliases
"""
user = g.user
alias: Alias = Alias.get(alias_id)
if alias.user_id != user.id:
return jsonify(error="Forbidden"), 403
return jsonify(**serialize_alias_info(get_alias_info(alias))), 200
2020-03-17 10:56:59 +01:00
def serialize_contact(fe: Contact) -> dict:
2020-03-14 12:22:43 +01:00
res = {
2020-03-17 12:38:50 +01:00
"id": fe.id,
2020-03-14 12:22:43 +01:00
"creation_date": fe.created_at.format(),
"creation_timestamp": fe.created_at.timestamp,
"last_email_sent_date": None,
"last_email_sent_timestamp": None,
"contact": fe.website_from or fe.website_email,
"reverse_alias": fe.website_send_to(),
}
2020-03-17 11:10:50 +01:00
fel: EmailLog = fe.last_reply()
2020-03-14 12:22:43 +01:00
if fel:
res["last_email_sent_date"] = fel.created_at.format()
res["last_email_sent_timestamp"] = fel.created_at.timestamp
return res
def get_alias_contacts(alias, page_id: int) -> [dict]:
2020-03-14 12:22:43 +01:00
q = (
2020-03-17 12:01:18 +01:00
Contact.query.filter_by(alias_id=alias.id)
2020-03-17 10:56:59 +01:00
.order_by(Contact.id.desc())
2020-03-14 12:22:43 +01:00
.limit(PAGE_LIMIT)
.offset(page_id * PAGE_LIMIT)
)
res = []
for fe in q.all():
2020-03-17 10:56:59 +01:00
res.append(serialize_contact(fe))
2020-03-14 12:22:43 +01:00
return res
@api_bp.route("/aliases/<int:alias_id>/contacts")
@cross_origin()
@verify_api_key
def get_alias_contacts_route(alias_id):
"""
Get alias contacts
Input:
page_id: in query
Output:
- contacts: list of contacts:
- creation_date
- creation_timestamp
- last_email_sent_date
- last_email_sent_timestamp
- contact
- reverse_alias
"""
user = g.user
try:
page_id = int(request.args.get("page_id"))
except (ValueError, TypeError):
return jsonify(error="page_id must be provided in request query"), 400
alias: Alias = Alias.get(alias_id)
2020-03-14 12:22:43 +01:00
if alias.user_id != user.id:
2020-03-14 12:22:43 +01:00
return jsonify(error="Forbidden"), 403
contacts = get_alias_contacts(alias, page_id)
2020-03-14 12:22:43 +01:00
return jsonify(contacts=contacts), 200
2020-03-14 12:55:38 +01:00
@api_bp.route("/aliases/<int:alias_id>/contacts", methods=["POST"])
@cross_origin()
@verify_api_key
def create_contact_route(alias_id):
"""
Create contact for an alias
Input:
alias_id: in url
contact: in body
Output:
201 if success
409 if contact already added
"""
data = request.get_json()
if not data:
return jsonify(error="request body cannot be empty"), 400
user = g.user
alias: Alias = Alias.get(alias_id)
2020-03-14 12:55:38 +01:00
if alias.user_id != user.id:
2020-03-14 12:55:38 +01:00
return jsonify(error="Forbidden"), 403
contact_addr = data.get("contact")
2020-03-14 12:55:38 +01:00
# generate a reply_email, make sure it is unique
# not use while to avoid infinite loop
reply_email = f"ra+{random_string(25)}@{EMAIL_DOMAIN}"
for _ in range(1000):
reply_email = f"ra+{random_string(25)}@{EMAIL_DOMAIN}"
2020-03-17 10:56:59 +01:00
if not Contact.get_by(reply_email=reply_email):
2020-03-14 12:55:38 +01:00
break
_, contact_email = parseaddr(contact_addr)
contact_email = contact_email.lower()
2020-03-14 12:55:38 +01:00
# already been added
if Contact.get_by(alias_id=alias.id, website_email=contact_email):
2020-03-14 12:55:38 +01:00
return jsonify(error="Contact already added"), 409
2020-03-17 10:56:59 +01:00
contact = Contact.create(
user_id=alias.user_id,
2020-03-17 12:01:18 +01:00
alias_id=alias.id,
website_email=contact_email,
website_from=contact_addr,
2020-03-14 12:55:38 +01:00
reply_email=reply_email,
)
LOG.d("create reverse-alias for %s %s", contact_addr, alias)
2020-03-14 12:55:38 +01:00
db.session.commit()
2020-03-17 10:56:59 +01:00
return jsonify(**serialize_contact(contact)), 201
2020-03-17 19:18:26 +01:00
@api_bp.route("/contacts/<int:contact_id>", methods=["DELETE"])
@cross_origin()
@verify_api_key
def delete_contact(contact_id):
"""
Delete contact
Input:
contact_id: in url
Output:
200
"""
user = g.user
contact = Contact.get(contact_id)
if not contact or contact.alias.user_id != user.id:
return jsonify(error="Forbidden"), 403
Contact.delete(contact_id)
db.session.commit()
return jsonify(deleted=True), 200