first commit

core/auth.py

@@ -0,0 +1,58 @@
+import core.users as users
+import core.postgres as postgres
+import bcrypt
+import jwt
+from jwt.exceptions import ExpiredSignatureError, InvalidTokenError
+import datetime
+import os
+
+
+def verifyLoginToken(login_token, username=False, userUUID=False):
+    if username:
+        userUUID = users.getUserUUID(username)
+
+    if userUUID:
+        try:
+            decoded_token = jwt.decode(
+                login_token, os.getenv("JWT_SECRET"), algorithms=["HS256"]
+            )
+            if decoded_token.get("sub") == str(userUUID):
+                return True
+            return False
+        except (ExpiredSignatureError, InvalidTokenError):
+            return False
+    return False
+
+
+def getUserpasswordHash(userUUID):
+    user = postgres.select_one("users", {"id": userUUID})
+    if user:
+        pw_hash = user.get("password_hashed")
+        if isinstance(pw_hash, memoryview):
+            return bytes(pw_hash)
+        return pw_hash
+    return None
+
+
+def getLoginToken(username, password):
+    userUUID = users.getUserUUID(username)
+    if userUUID:
+        formatted_pass = password.encode("utf-8")
+        users_hashed_pw = getUserpasswordHash(userUUID)
+        if bcrypt.checkpw(formatted_pass, users_hashed_pw):
+            payload = {
+                "sub": userUUID,
+                "name": users.getUserFirstName(userUUID),
+                "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1),
+            }
+            return jwt.encode(payload, os.getenv("JWT_SECRET"), algorithm="HS256")
+    return False
+
+
+def unregisterUser(userUUID, password):
+    pw_hash = getUserpasswordHash(userUUID)
+    if not pw_hash:
+        return False
+    if bcrypt.checkpw(password.encode("utf-8"), pw_hash):
+        return users.deleteUser(userUUID)
+    return False