Merge pull request #74 from simple-login/api-alias2

Add some alias endpoints for API

README.md

@@ -758,7 +758,8 @@ If success, 200 with the list of aliases, for example:
             "id": 4,
             "nb_block": 0,
             "nb_forward": 0,
-            "nb_reply": 0
+            "nb_reply": 0,
+            "enabled": true 
         },
         {
             "creation_date": "2020-02-04 16:23:02+00:00",
@@ -767,12 +768,79 @@ If success, 200 with the list of aliases, for example:
             "id": 3,
             "nb_block": 0,
             "nb_forward": 0,
-            "nb_reply": 0
+            "nb_reply": 0,
+            "enabled": false
         }
     ]
 }
 ```
 
+#### DELETE /api/aliases/:alias_id
+
+Delete an alias
+
+Input:
+- `Authentication` header that contains the api key
+- `alias_id` in url. 
+
+Output:
+If success, 200.
+
+
+```json
+{
+    "deleted": true
+}
+```
+
+#### POST /api/aliases/:alias_id/toggle
+
+Enable/disable alias
+
+Input:
+- `Authentication` header that contains the api key
+- `alias_id` in url. 
+
+Output:
+If success, 200 along with the new alias status:
+
+```json
+{
+    "enabled": false
+}
+```
+
+#### GET /api/aliases/:alias_id/activities
+
+Get activities for a given alias.
+
+Input:
+- `Authentication` header that contains the api key
+- `alias_id`: the alias id, passed in url.
+- `page_id` used in request query (`?page_id=0`). The endpoint returns maximum 20 aliases for each page. `page_id` starts at 0.
+
+Output:
+If success, 200 with the list of activities, for example:
+
+```json
+{
+  "activities": [
+    {
+      "action": "reply",
+      "from": "yes_meo_chat@sl.local",
+      "timestamp": 1580903760,
+      "to": "marketing@example.com"
+    },
+    {
+      "action": "reply",
+      "from": "yes_meo_chat@sl.local",
+      "timestamp": 1580903760,
+      "to": "marketing@example.com"
+    }
+  ]
+}
+```
+
 
 ### Database migration
 

app/api/views/alias.py

@@ -3,13 +3,10 @@ from flask import jsonify, request
 from flask_cors import cross_origin
 
 from app.api.base import api_bp, verify_api_key
-from app.config import MAX_NB_EMAIL_FREE_PLAN
-from app.dashboard.views.custom_alias import verify_prefix_suffix
+from app.dashboard.views.alias_log import get_alias_log
 from app.dashboard.views.index import get_alias_info, AliasInfo
 from app.extensions import db
-from app.log import LOG
-from app.models import GenEmail, AliasUsedOn
-from app.utils import convert_to_id
+from app.models import GenEmail
 
 
 @api_bp.route("/aliases")
@@ -50,9 +47,113 @@ def get_aliases():
                     "nb_forward": alias.nb_forward,
                     "nb_block": alias.nb_blocked,
                     "nb_reply": alias.nb_reply,
+                    "enabled": alias.gen_email.enabled,
                 }
                 for alias in aliases
             ]
         ),
         200,
     )
+
+
+@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
+    gen_email = GenEmail.get(alias_id)
+
+    if gen_email.user_id != user.id:
+        return jsonify(error="Forbidden"), 403
+
+    GenEmail.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
+    gen_email: GenEmail = GenEmail.get(alias_id)
+
+    if gen_email.user_id != user.id:
+        return jsonify(error="Forbidden"), 403
+
+    gen_email.enabled = not gen_email.enabled
+    db.session.commit()
+
+    return jsonify(enabled=gen_email.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
+
+    """
+    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
+
+    gen_email: GenEmail = GenEmail.get(alias_id)
+
+    if gen_email.user_id != user.id:
+        return jsonify(error="Forbidden"), 403
+
+    alias_logs = get_alias_log(gen_email, page_id)
+
+    activities = []
+    for alias_log in alias_logs:
+        activity = {"timestamp": alias_log.when.timestamp}
+        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
+
+            if alias_log.blocked:
+                activity["action"] = "block"
+            else:
+                activity["action"] = "forward"
+
+        activities.append(activity)
+
+    return (
+        jsonify(activities=activities),
+        200,
+    )

app/api/views/auth_login.py

@@ -56,8 +56,11 @@ def auth_login():
         ret["mfa_key"] = s.sign(str(user.id))
         ret["api_key"] = None
     else:
-        api_key = ApiKey.create(user.id, device)
-        db.session.commit()
+        api_key = ApiKey.get_by(user_id=user.id, name=device)
+        if not api_key:
+            LOG.d("create new api key for %s and %s", user, device)
+            api_key = ApiKey.create(user.id, device)
+            db.session.commit()
         ret["mfa_key"] = None
         ret["api_key"] = api_key.code
 

app/api/views/auth_mfa.py

@@ -6,6 +6,7 @@ from itsdangerous import Signer, BadSignature
 from app.api.base import api_bp
 from app.config import FLASK_SECRET
 from app.extensions import db
+from app.log import LOG
 from app.models import User, ApiKey
 
 
@@ -58,8 +59,12 @@ def auth_mfa():
         "name": user.name,
     }
 
-    api_key = ApiKey.create(user.id, device)
-    db.session.commit()
+    api_key = ApiKey.get_by(user_id=user.id, name=device)
+    if not api_key:
+        LOG.d("create new api key for %s and %s", user, device)
+        api_key = ApiKey.create(user.id, device)
+        db.session.commit()
+
     ret["api_key"] = api_key.code
 
     return jsonify(**ret), 200

tests/api/test_alias.py

@@ -1,8 +1,10 @@
+import json
+
 from flask import url_for
 
 from app.config import EMAIL_DOMAIN, MAX_NB_EMAIL_FREE_PLAN, PAGE_LIMIT
 from app.extensions import db
-from app.models import User, ApiKey, GenEmail
+from app.models import User, ApiKey, GenEmail, ForwardEmail, ForwardEmailLog
 from app.utils import random_word
 
 
@@ -54,3 +56,95 @@ def test_success_with_pagination(flask_client):
     )
     assert r.status_code == 200
     assert len(r.json["aliases"]) == 2
+
+
+def test_delete_alias(flask_client):
+    user = User.create(
+        email="a@b.c", password="password", name="Test User", activated=True
+    )
+    db.session.commit()
+
+    # create api_key
+    api_key = ApiKey.create(user.id, "for test")
+    db.session.commit()
+
+    gen_email = GenEmail.create_new_random(user.id)
+    db.session.commit()
+
+    r = flask_client.delete(
+        url_for("api.delete_alias", alias_id=gen_email.id),
+        headers={"Authentication": api_key.code},
+    )
+
+    assert r.status_code == 200
+    assert r.json == {"deleted": True}
+
+
+def test_toggle_alias(flask_client):
+    user = User.create(
+        email="a@b.c", password="password", name="Test User", activated=True
+    )
+    db.session.commit()
+
+    # create api_key
+    api_key = ApiKey.create(user.id, "for test")
+    db.session.commit()
+
+    gen_email = GenEmail.create_new_random(user.id)
+    db.session.commit()
+
+    r = flask_client.post(
+        url_for("api.toggle_alias", alias_id=gen_email.id),
+        headers={"Authentication": api_key.code},
+    )
+
+    assert r.status_code == 200
+    assert r.json == {"enabled": False}
+
+
+def test_alias_activities(flask_client):
+    user = User.create(
+        email="a@b.c", password="password", name="Test User", activated=True
+    )
+    db.session.commit()
+
+    # create api_key
+    api_key = ApiKey.create(user.id, "for test")
+    db.session.commit()
+
+    gen_email = GenEmail.create_new_random(user.id)
+    db.session.commit()
+
+    # create some alias log
+    forward_email = ForwardEmail.create(
+        website_email="marketing@example.com",
+        reply_email="reply@a.b",
+        gen_email_id=gen_email.id,
+    )
+    db.session.commit()
+
+    for _ in range(int(PAGE_LIMIT / 2)):
+        ForwardEmailLog.create(forward_id=forward_email.id, is_reply=True)
+
+    for _ in range(int(PAGE_LIMIT / 2) + 2):
+        ForwardEmailLog.create(forward_id=forward_email.id, blocked=True)
+
+    r = flask_client.get(
+        url_for("api.get_alias_activities", alias_id=gen_email.id, page_id=0),
+        headers={"Authentication": api_key.code},
+    )
+
+    assert r.status_code == 200
+    assert len(r.json["activities"]) == PAGE_LIMIT
+    for ac in r.json["activities"]:
+        assert ac["action"]
+        assert ac["from"]
+        assert ac["action"]
+        assert ac["action"]
+
+    # second page, should return 1 or 2 results only
+    r = flask_client.get(
+        url_for("api.get_alias_activities", alias_id=gen_email.id, page_id=1),
+        headers={"Authentication": api_key.code},
+    )
+    assert len(r.json["activities"]) < 3

tests/api/test_auth_login.py

@@ -40,3 +40,26 @@ def test_auth_login_success_mfa_enabled(flask_client):
     assert r.json["mfa_enabled"] == True
     assert r.json["mfa_key"]
     assert r.json["name"] == "Test User"
+
+
+def test_auth_login_device_exist(flask_client):
+    User.create(email="a@b.c", password="password", name="Test User", activated=True)
+    db.session.commit()
+
+    r = flask_client.post(
+        url_for("api.auth_login"),
+        json={"email": "a@b.c", "password": "password", "device": "Test Device"},
+    )
+
+    assert r.status_code == 200
+    api_key = r.json["api_key"]
+    assert r.json["mfa_enabled"] == False
+    assert r.json["mfa_key"] is None
+    assert r.json["name"] == "Test User"
+
+    # same device, should return same api_key
+    r = flask_client.post(
+        url_for("api.auth_login"),
+        json={"email": "a@b.c", "password": "password", "device": "Test Device"},
+    )
+    assert r.json["api_key"] == api_key