feat: hashing user passwords

Author: chanh-1

requirements.txt

@@ -5,4 +5,5 @@ sqlalchemy
 pyjwt[crypto]
 jinja2
 langflow==0.0.54
-black
+black
+passlib

server/api/auth.py

@@ -2,6 +2,7 @@
 import database
 from database import User
 from typing import Annotated
+from passlib.hash import sha256_crypt
 from sqlalchemy.orm import Session
 from fastapi import APIRouter, Depends, Query, Header
 from pydantic import BaseModel
@@ -18,8 +19,8 @@ class AuthModel(BaseModel):
 
 @auth_router.post("/login", status_code=200)
 def login(auth: AuthModel, db: Session = Depends(database.db_session)):
-    user: User = db.query(User).filter((User.username == auth.username) & (User.password == auth.password)).first()  # type: ignore
-    if user is not None:
+    user: User = db.query(User).filter(User.username == auth.username).first()  # type: ignore
+    if user is not None and sha256_crypt.verify(auth.password, user.password):  # type: ignore
         token = jwt.encode(payload={"username": auth.username}, key=JWT_SECRET)
         response = {"msg": "success", "token": token}
     else:

server/api/user.py

@@ -2,6 +2,7 @@
 from typing import Annotated
 from database import User
 from sqlalchemy.orm import Session
+from passlib.hash import sha256_crypt
 from fastapi import APIRouter, Depends, Query, Header
 from pydantic import BaseModel
 from commons.utils import get_user_from_jwt, verify_user
@@ -17,13 +18,14 @@ class ChangePasswordModel(BaseModel):
 
 @user_router.post("/change_password", status_code=200)
 def change_password(inputs: ChangePasswordModel, token: Annotated[str, Header()], db: Session = Depends(database.db_session)):
-    user = get_user_from_jwt(token)
-    verify_user(user)
-    user: User = db.query(User).filter((User.username == inputs.username) & (User.password == inputs.old_password)).first()
-    if user is not None:
-        user.password = inputs.new_password
+    username = get_user_from_jwt(token)
+    verify_user(username)
+    user: User = db.query(User).filter(User.username == inputs.username).first()  # type: ignore
+    if sha256_crypt.verify(inputs.old_password, user.password):  # type: ignore
+        password = sha256_crypt.hash(inputs.new_password)
+        user.password = password  # type: ignore
         db.commit()
         response = {"msg": "success"}
     else:
-        response = {"msg": "failed"}
+        response = 400, {"msg": "You have entered the wrong password"}
     return response

server/commons/utils.py

@@ -1,4 +1,5 @@
 import jwt
+from passlib.hash import sha256_crypt
 from sqlalchemy.exc import IntegrityError
 from datetime import datetime, timedelta
 from database import db_session, User, Prompt, IntermediateStep
@@ -9,7 +10,8 @@
 
 
 def add_default_user():
-    new_user = User(username="admin", password="admin", meta="")
+    admin_password = sha256_crypt.hash("admin")
+    new_user = User(username="admin", password=admin_password, meta="")
     db = db_session()
     try:
         db.add(new_user)