diff --git a/server/app/apis/auth.py b/server/app/apis/auth.py
index bf9eeefde781f3fcacfd4bdafade8829db24acaa..37b505a6eb9dee4e8f625180b66bdfd02ed92449 100644
--- a/server/app/apis/auth.py
+++ b/server/app/apis/auth.py
@@ -1,4 +1,4 @@
-from datetime import timedelta
+from datetime import datetime, timedelta
import app.core.http_codes as codes
import app.database.controller as dbc
@@ -7,6 +7,7 @@ from app.core import sockets
from app.core.codes import verify_code
from app.core.dto import AuthDTO
from app.database.models import Whitelist
+from flask import current_app
from flask_jwt_extended import create_access_token, get_jti, get_raw_jwt
from flask_jwt_extended.utils import get_jti
from flask_restx import Resource, inputs, reqparse
@@ -26,6 +27,9 @@ create_user_parser.add_argument("role_id", type=int, required=True, location="js
login_code_parser = reqparse.RequestParser()
login_code_parser.add_argument("code", type=str, required=True, location="json")
+USER_LOGIN_LOCKED_ATTEMPTS = current_app.config["USER_LOGIN_LOCKED_ATTEMPTS"]
+USER_LOGIN_LOCKED_EXPIRES = current_app.config["USER_LOGIN_LOCKED_EXPIRES"]
+
def get_user_claims(item_user):
return {"role": item_user.role.name, "city_id": item_user.city_id}
@@ -54,9 +58,11 @@ class AuthSignup(Resource):
args = create_user_parser.parse_args(strict=True)
email = args.get("email")
+ # Check if email is already used
if dbc.get.user_exists(email):
api.abort(codes.BAD_REQUEST, "User already exists")
+ # Add user
item_user = dbc.add.user(**args)
return item_response(schema.dump(item_user))
@@ -67,9 +73,12 @@ class AuthDelete(Resource):
@protect_route(allowed_roles=["Admin"])
def delete(self, user_id):
item_user = dbc.get.user(user_id)
+
+ # Blacklist all the whitelisted tokens in use for the user that will be deleted
dbc.delete.whitelist_to_blacklist(Whitelist.user_id == user_id)
- dbc.delete.default(item_user)
+ # Delete user
+ dbc.delete.default(item_user)
return text_response(f"User {user_id} deleted")
@@ -79,15 +88,44 @@ class AuthLogin(Resource):
args = login_parser.parse_args(strict=True)
email = args.get("email")
password = args.get("password")
+
item_user = dbc.get.user_by_email(email)
- if not item_user or not item_user.is_correct_password(password):
+ # Login with unkown email
+ if not item_user:
+ api.abort(codes.UNAUTHORIZED, "Invalid email or password")
+
+ # Login with existing email but with wrong password
+ if not item_user.is_correct_password(password):
+ # Increase the login attempts every time the user tries to login with wrong password
+ item_user.login_attempts += 1
+
+ # Lock the user out for some time
+ if item_user.login_attempts == USER_LOGIN_LOCKED_ATTEMPTS:
+ item_user.locked = datetime.now() + USER_LOGIN_LOCKED_EXPIRES
+
+ dbc.utils.commit()
api.abort(codes.UNAUTHORIZED, "Invalid email or password")
+ # Otherwise if login was successful but the user is locked
+ if item_user.locked:
+ # Check if locked is greater than now
+ if item_user.locked > datetime.now():
+ api.abort(codes.UNAUTHORIZED, f"Try again in {item_user.locked} hours.")
+ else:
+ item_user.locked = None
+
+ # If everything else was successful, set login_attempts to 0
+ item_user.login_attempts = 0
+ dbc.utils.commit()
+
+ # Create the jwt with user.id as the identifier
access_token = create_access_token(item_user.id, user_claims=get_user_claims(item_user))
- # refresh_token = create_refresh_token(item_user.id)
+ # Login response includes the id and jwt for the user
response = {"id": item_user.id, "access_token": access_token}
+
+ # Whitelist the created jwt
dbc.add.whitelist(get_jti(access_token), item_user.id)
return response
@@ -98,6 +136,7 @@ class AuthLoginCode(Resource):
args = login_code_parser.parse_args()
code = args["code"]
+ # Check so the code string is valid
if not verify_code(code):
api.abort(codes.UNAUTHORIZED, "Invalid code")
@@ -107,10 +146,12 @@ class AuthLoginCode(Resource):
if item_code.competition_id not in sockets.presentations:
api.abort(codes.UNAUTHORIZED, "Competition not active")
+ # Create jwt that is only valid for 8 hours
access_token = create_access_token(
item_code.id, user_claims=get_code_claims(item_code), expires_delta=timedelta(hours=8)
)
+ # Whitelist the created jwt
dbc.add.whitelist(get_jti(access_token), competition_id=item_code.competition_id)
response = {
"competition_id": item_code.competition_id,
@@ -126,8 +167,13 @@ class AuthLogout(Resource):
@protect_route(allowed_roles=["*"], allowed_views=["*"])
def post(self):
jti = get_raw_jwt()["jti"]
+
+ # Blacklist the token so the user cannot access the api anymore
dbc.add.blacklist(jti)
+
+ # Remove the the token from the whitelist since it's blacklisted now
Whitelist.query.filter(Whitelist.jti == jti).delete()
+
dbc.utils.commit()
return text_response("Logout")
diff --git a/server/app/database/models.py b/server/app/database/models.py
index a54a77c24b6b77a22187c386e8cd59c1cd006f49..97eb6097c2403cf4d32d01106e2ae906d9d788d5 100644
--- a/server/app/database/models.py
+++ b/server/app/database/models.py
@@ -61,8 +61,9 @@ class User(db.Model):
_password = db.Column(db.LargeBinary(60), nullable=False)
authenticated = db.Column(db.Boolean, default=False)
- # twoAuthConfirmed = db.Column(db.Boolean, default=True)
- # twoAuthCode = db.Column(db.String(STRING_SIZE), nullable=True)
+
+ login_attempts = db.Column(db.Integer, nullable=False, default=0)
+ locked = db.Column(db.DateTime(timezone=True), nullable=True, default=None)
role_id = db.Column(db.Integer, db.ForeignKey("role.id"), nullable=False)
city_id = db.Column(db.Integer, db.ForeignKey("city.id"), nullable=False)
diff --git a/server/configmodule.py b/server/configmodule.py
index 93d21cbe84a5847ed4927a4d2b3d002b487d1f69..e38c4e04cf1b90ffff4edbe0cbcffffb347c5b4a 100644
--- a/server/configmodule.py
+++ b/server/configmodule.py
@@ -17,6 +17,8 @@ class Config:
THUMBNAIL_SIZE = (120, 120)
SECRET_KEY = os.urandom(24)
SQLALCHEMY_ECHO = False
+ USER_LOGIN_LOCKED_ATTEMPTS = 12
+ USER_LOGIN_LOCKED_EXPIRES = timedelta(hours=3)
class DevelopmentConfig(Config):
@@ -34,6 +36,8 @@ class DevelopmentConfig(Config):
class TestingConfig(Config):
TESTING = True
SQLALCHEMY_DATABASE_URI = "sqlite:///test.db"
+ USER_LOGIN_LOCKED_ATTEMPTS = 4
+ USER_LOGIN_LOCKED_EXPIRES = timedelta(seconds=4)
class ProductionConfig(Config):
diff --git a/server/tests/test_app.py b/server/tests/test_app.py
index 2152a6d4c7236f57d099f50dccec3a090ce94c5d..880a41bd82a3b6cac8795a3cf61eaed6569dbadb 100644
--- a/server/tests/test_app.py
+++ b/server/tests/test_app.py
@@ -2,15 +2,37 @@
This file tests the api function calls.
"""
+import time
+
import app.core.http_codes as codes
-from app.database.controller.add import competition
-from app.database.models import Slide
+import pytest
from app.core import sockets
from tests import app, client, db
from tests.test_helpers import add_default_values, change_order_test, delete, get, post, put
+# @pytest.mark.skip(reason="Takes long time")
+def test_locked_api(client):
+ add_default_values()
+
+ # Login in with default user but wrong password until blocked
+ for i in range(4):
+ response, body = post(client, "/api/auth/login", {"email": "test@test.se", "password": "password1"})
+ assert response.status_code == codes.UNAUTHORIZED
+
+ # Login with right password, user should be locked
+ response, body = post(client, "/api/auth/login", {"email": "test@test.se", "password": "password"})
+ assert response.status_code == codes.UNAUTHORIZED
+
+ # Sleep for 4 secounds
+ time.sleep(4)
+
+ # Check so the user is no longer locked
+ response, body = post(client, "/api/auth/login", {"email": "test@test.se", "password": "password"})
+ assert response.status_code == codes.OK
+
+
def test_misc_api(client):
add_default_values()
@@ -125,6 +147,10 @@ def test_auth_and_user_api(client):
assert response.status_code == codes.OK
headers = {"Authorization": "Bearer " + body["access_token"]}
+ # Login in with default user but wrong password
+ response, body = post(client, "/api/auth/login", {"email": "test@test.se", "password": "password1"})
+ assert response.status_code == codes.UNAUTHORIZED
+
# Create user
register_data = {"email": "test1@test.se", "password": "abc123", "role_id": 2, "city_id": 1}
response, body = post(client, "/api/auth/signup", register_data, headers)
@@ -211,7 +237,6 @@ def test_auth_and_user_api(client):
assert response.status_code == codes.OK
# TODO: Check if current users jwt (jti) is in blacklist after logging out
-
response, body = get(client, "/api/users", headers=headers)
assert response.status_code == codes.UNAUTHORIZED
@@ -479,4 +504,4 @@ def test_authorization(client):
# Also get antoher teams answers
response, body = get(client, f"/api/competitions/{competition_id}/teams/{team_id+1}/answers", headers=headers)
- assert response.status_code == codes.OK
\ No newline at end of file
+ assert response.status_code == codes.OK