Activate new admin and superadmin accounts by mail

.env.example

@@ -4,12 +4,16 @@ DATABASE_URL="sqlite:///./db.sqlite"
 # AUTH SETTINGS / PASSWORT RESET
 # Important: No Backslash at the end!
 PASSWORD_RESET_LINK="127.0.0.1"
+
+ACTIVATE_ACCOUNT_LINK="127.0.0.1"
 #In Sec = 20min 
 JWT_VALID_TIME_ACCESS=1200 
 #In Sec = one week
 JWT_VALID_TIME_REFRESH=604800 
 #In Sec = 10min
 JWT_VALID_TIME_PWD_RESET=600 
+#In Sec = 10min
+JWT_VALID_TIME_ACTIVATE_ACCOUNT=1800
 JWT_SECRET="secret"
 JWT_ALGORITHM="HS256"
 MAIL_USERNAME="[email protected]"

.github/workflows/dev.yml

@@ -41,6 +41,7 @@ jobs:
           echo "JWT_VALID_TIME_ACCESS=1200" >> .env
           echo "JWT_VALID_TIME_REFRESH=604800" >> .env
           echo "JWT_VALID_TIME_PWD_RESET=600" >> .env
+          echo "JWT_VALID_TIME_ACTIVATE_ACCOUNT=1800" >> .env
           echo "JWT_ALGORITHM=HS256" >> .env
           echo "[email protected]" >> .env
           echo "MAIL_PORT=587" >> .env
@@ -49,6 +50,7 @@ jobs:
           echo "[email protected]" >> .env
           echo "SUPERADMIN_PASSWORD={{ secrets.DEV_SUPERADMIN_PASSWORD }}" >> .env
           echo "PASSWORD_RESET_LINK=https://admin-kosjenka-dev.vercel.app/password/confirm" >> .env
+          echo "ACTIVATE_ACCOUNT_LINK=https://admin-kosjenka-dev.vercel.app/password/confirm" >> .env
 
       - name: Build and push Docker image
         uses: docker/build-push-action@ad44023a93711e3deb337508980b4b5e9bcdc5dc

.github/workflows/pipeline.yml

@@ -8,9 +8,11 @@ on:
 env:
   DATABASE_URL: "sqlite:///./db.sqlite"
   PASSWORD_RESET_LINK: "127.0.0.1" 
+  ACTIVATE_ACCOUNT_LINK: "127.0.0.1:8000"
   JWT_VALID_TIME_ACCESS: 1200 
   JWT_VALID_TIME_REFRESH: 604800
   JWT_VALID_TIME_PWD_RESET: 600 
+  JWT_VALID_TIME_ACTIVATE_ACCOUNT: 1800
   JWT_SECRET: "secret"
   JWT_ALGORITHM: "HS256"
   MAIL_USERNAME: "[email protected]"

.github/workflows/prod.yml

@@ -13,6 +13,7 @@ env:
   JWT_VALID_TIME_ACCESS: 1200 
   JWT_VALID_TIME_REFRESH: 604800
   JWT_VALID_TIME_PWD_RESET: 600 
+  JWT_VALID_TIME_ACTIVATE_ACCOUNT: 1800
   JWT_ALGORITHM: "HS256"
   MAIL_USERNAME: "[email protected]"
   MAIL_PORT: 587
@@ -59,6 +60,7 @@ jobs:
           echo "MAIL_FROM_NAME=Kosjenka Support" >> .env
           echo "SUPERADMIN_PASSWORD=${{ secrets.PROD_SUPERADMIN_PASSWORD }}" >> .env
           echo "PASSWORD_RESET_LINK=https://admin-kosjenka.vercel.app/password/confirm" >> .env
+          echo "ACTIVATE_ACCOUNT_LINK=echo "PASSWORD_RESET_LINK=https://admin-kosjenka.vercel.app/password/confirm" >> .env
 
       - name: Build and push Docker image
         uses: docker/build-push-action@ad44023a93711e3deb337508980b4b5e9bcdc5dc

auth.py

@@ -18,6 +18,7 @@
 JWT_VALID_TIME_PWD_RESET = int(
     os.environ["JWT_VALID_TIME_PWD_RESET"]
 )  # 60 * 10  # 10min
+JWT_VALID_TIME_ACTIVATE_ACCOUNT = int(os.environ["JWT_VALID_TIME_ACTIVATE_ACCOUNT"])
 JWT_SECRET = os.environ["JWT_SECRET"]  # "C0ddVvlcaL4UuChF8ckFQoVCGbtizyvK"
 JWT_ALGORITHM = os.environ["JWT_ALGORITHM"]  # "HS256"
 
@@ -154,3 +155,48 @@ def reset_password(db: Session, new_password: str, token: str):
         return "SUCCESS"
     except:
         return "ERROR"
+
+
+# Admin Password set
+def create_account_activation_token(
+    email: EmailStr, is_superadmin: bool, valid_time: int
+):
+    payload = {
+        "email": email,
+        "is_superadmin": is_superadmin,
+        "expires": time.time() + valid_time,
+    }
+    token = jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
+    return token
+
+
+async def send_account_password_mail(account: schemas.AccountPostAdminIn):
+    token = create_account_activation_token(
+        email=account.email,
+        is_superadmin=account.is_superadmin,
+        valid_time=JWT_VALID_TIME_ACTIVATE_ACCOUNT,
+    )
+    link_base = os.environ["ACTIVATE_ACCOUNT_LINK"]
+    template_body = {
+        "user": account.email,
+        "url": f"{link_base}?token={token}",
+        "expire_in_minutes": int(JWT_VALID_TIME_ACTIVATE_ACCOUNT / 60),
+    }
+    message = MessageSchema(
+        subject="Kosjenka - Account Registration",
+        recipients=[account.email],
+        template_body=template_body,
+        subtype=MessageType.html,
+    )
+    fm = FastMail(conf)
+    await fm.send_message(message, template_name="activate_account_email.html")
+
+
+def check_account_activation_token(token: str):
+    try:
+        decoded_token = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
+        if decoded_token["expires"] < time.time():
+            return None
+        return decoded_token
+    except:
+        return None

html_templates/activate_account_email.html

@@ -0,0 +1,18 @@
+<!DOCTYPE html>
+<html>
+    <head>
+        <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
+        <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+        <title>Activate your account</title>
+    </head>
+    <body>
+    <h2 style="color: #27a9e1;"> Hey!</h2>
+    <p style="font-size:18px;color: #030303;">It seems a account for you was created!</p>
+    <p style="font-size:18px;color: #030303;">Follow this link to finish the registration and set a password for your account:</p>
+    <p><a style="font-size:18px;display: inline-block; padding: 10px 20px; background-color: #27a9e1; color: #ffffff; text-decoration: none; border-radius: 5px;" href="{{url}}">Set password</a></p>
+    <b style="font-size:18px;color: #ff9999;">Note: This link will expire in {{expire_in_minutes}} minutes.</b>
+    <br>
+    <hr style="border: 1px solid #CCCCCC;">
+    <p style="font-size:14px;color: #888888;">If you've received this mail without filling a reqeust, it's likely that a user entered your email address by mistake. In that case, just ignore it.</p>
+    </body>
+</html>

main.py

@@ -218,21 +218,42 @@ def track_exercise_completion(
     return db_do_exercise
 
 
-@app.post("/accounts", response_model=schemas.AccountOut)
-def create_account(
-    account_in: schemas.AccountPostAdmin,
+@app.post("/accounts")
+async def create_account(
+    account_in: schemas.AccountPostAdminIn,
     db: Session = Depends(get_db),
     auth_user: schemas.AuthSchema = Depends(JWTBearer()),
 ):
     validate_access_level(auth_user, models.AccountType.Superadmin)
-    if crud.email_is_registered(db, account_in.email):
-        raise HTTPException(status_code=409, detail="Email already registered")
 
-    if account_in.is_superadmin:
+    try:
+        await auth.send_account_password_mail(account=account_in)
+        return {
+            "result": f"An email has been sent to {account_in.email} with a link for activating the account."
+        }
+    except Exception as e:
+        print(e)
+        raise HTTPException(status_code=500, detail=f"An unexpected error occurred")
+
+
+@app.post("/accounts/activate", response_model=schemas.AccountOut)
+def account_reset_password_result(
+    input: schemas.ActivateAccountSchema,
+    db: Session = Depends(get_db),
+):
+    result = auth.check_account_activation_token(input.token)
+    if result == None:
+        raise HTTPException(status_code=401, detail="Token is expired or not valid")
+    if crud.email_is_registered(db, result["email"]):
+        raise HTTPException(status_code=409, detail="Email already registered")
+    if result["is_superadmin"]:
         type_account = models.AccountType.Superadmin
     else:
         type_account = models.AccountType.Admin
-    account_saved = crud.create_account(db, account_in, type_account)
+    new_account = schemas.AccountPostAdmin
+    new_account.email = result["email"]
+    new_account.password = input.password
+    account_saved = crud.create_account(db, new_account, type_account)
     return account_saved
 
 

schemas.py

@@ -81,6 +81,11 @@ class AccountIn(BaseModel):
     password: str
 
 
+class AccountPostAdminIn(BaseModel):
+    email: EmailStr
+    is_superadmin: Optional[bool] = False
+
+
 class AccountPostAdmin(AccountIn):
     is_superadmin: Optional[bool] = False
 
@@ -147,3 +152,9 @@ class ResetPasswordSchema(BaseModel):
 
 class ResetPasswordResultSchema(BaseModel):
     details: str
+
+
+# Activate Account
+class ActivateAccountSchema(BaseModel):
+    password: str
+    token: str

test/conftest.py

@@ -4,6 +4,7 @@
 from crud import password_hasher
 from database import SessionLocal
 from main import app
+from auth import create_account_activation_token
 import models
 
 
@@ -26,14 +27,16 @@ def regular_token():
 @pytest.fixture(scope="session")
 def admin_token(superadmin_token):
     account_details = {"email": "[email protected]", "password": "admin"}
+    activate = {
+        "token": create_account_activation_token("[email protected]", False, 60000),
+        "password": "admin",
+    }
     resp = client.post("http://localhost:8000/login", json=account_details).json()
     if "detail" in resp and resp["detail"] == "Username/Password wrong":
         resp = client.post(
-            "http://localhost:8000/accounts",
-            json=account_details,
-            headers=auth_header(superadmin_token),
+            "http://localhost:8000/accounts/activate", json=activate
         ).json()
-        resp = client.post("http://localhost:8000/login", json=account_details).json()
+    resp = client.post("http://localhost:8000/login", json=account_details).json()
     access_token = resp["access_token"]
     yield access_token
 

test/test_accounts.py

@@ -1,5 +1,7 @@
 from conftest import client, auth_header, good_request, bad_request
 
+from auth import createPasswortResetToken, create_account_activation_token
+
 
 def test_create_account(superadmin_token):
     accounts = client.get(
@@ -13,12 +15,38 @@ def test_create_account(superadmin_token):
         headers=auth_header(superadmin_token),
     )
     assert resp.status_code == 200
+    activate = {
+        "token": create_account_activation_token("[email protected]", False, 6000),
+        "password": "secret",
+    }
+    activate_result = client.post(
+        "http://localhost:8000/accounts/activate", json=activate
+    )
+    assert activate_result.status_code == 200
     accounts = client.get(
         "http://localhost:8000/accounts", headers=auth_header(superadmin_token)
     ).json()
     assert len(accounts["items"]) == account_count + 1
 
 
+def test_activate_account():
+    account = {"email": "[email protected]", "password": "secret"}
+    resp = client.post("http://localhost:8000/login", json=account)
+    # Check if login is not possible
+    assert resp.status_code == 400
+    token = create_account_activation_token("[email protected]", False, 6000)
+    activate = {
+        "token": token,
+        "password": "secret",
+    }
+
+    resp2 = client.post("http://localhost:8000/accounts/activate", json=activate)
+    assert resp2.status_code == 200
+    # Try to login again
+    resp3 = client.post("http://localhost:8000/login", json=account)
+    assert resp3.status_code == 200
+
+
 def test_update_account(superadmin_token):
     # Get the superadmin
     accounts = client.get(