✨️ User Creation/Login Update

Author: pheeef

.gitignore

@@ -12,6 +12,7 @@ __pycache__/
 # Distribution / packaging
 .Python
 build/
+testing/
 develop-eggs/
 dist/
 downloads/
@@ -130,3 +131,4 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+/testing/

assets/database.py

@@ -1,15 +1,15 @@
 import mysql.connector
-from env import *
+import env as e
 # env (enviorment) is the env.py file where all of the
 # variables are stored for the database access create
 # your own file with all the needed variables (see below)
 
 
 def openDBConnection():
     db = mysql.connector.connect(
-        host=host,
-        user=username,
-        passwd=password,
-        database=database
+        host=e.host,
+        user=e.username,
+        passwd=e.password,
+        database=e.database
     )
     return db

functions/hashing.py

@@ -1,16 +1,74 @@
+from fastapi import Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+from jose import jwt, JWTError
 from passlib.context import CryptContext
+from starlette.status import HTTP_401_UNAUTHORIZED
+
+from functions import user as u
+import env as e
+from functions.user import get_user
+from models.tokendata import TokenData
+from models.user import User
 
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
-oauth2_sheme = OAuth2PasswordBearer(tokenUrl="token")
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
-SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
 ALOGRITHM = "HS256"
 ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
+
 def verify_password(plain_password, hashed_password):
     return pwd_context.verify(plain_password, hashed_password)
 
 
 def get_password_hash(password):
-    return pwd_context.hash(password)
+    return pwd_context.hash(password)
+
+
+def authenticate_user(EMAIL: str, PASSWORD: str):
+    account = u.get_user(EMAIL=EMAIL)
+    if not account:
+        return False
+    if not verify_password(PASSWORD, account.PASSWORD_HASH):
+        return False
+    return account
+
+
+async def get_current_user(token: str = Depends(oauth2_scheme)):
+    credential_email_none_exception = HTTPException(
+        status_code=HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials [email none]",
+        headers={"WWW-Authenticate": "Bearer"}
+    )
+    credential_jwt_exception = HTTPException(
+        status_code=HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials [jwt]",
+        headers={"WWW-Authenticate": "Bearer"}
+    )
+    credential_email_db_exception = HTTPException(
+        status_code=HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials [email-db]",
+        headers={"WWW-Authenticate": "Bearer"}
+    )
+    try:
+        # dekodieren des JWT
+        payload = jwt.decode(token, e.SECRET_KEY, algorithms=[ALOGRITHM])
+        # get EMAIL out to the JWT package
+        EMAIL: str = payload.get("sub")
+        # check if field EMAIL is set
+        if EMAIL is None:
+            # if not raise credential exception
+            raise credential_email_none_exception
+        token_data = TokenData(EMAIL=EMAIL)
+    except JWTError:
+        raise credential_jwt_exception
+    user = get_user(EMAIL=token_data.EMAIL)
+    if user is None:
+        raise credential_email_db_exception
+    return user
+
+
+async def get_current_active_user(current_user: User = Depends(get_current_user)):
+    if not current_user.ACTIVE:
+        raise HTTPException(status_code=400, detail="Inactive user")
+    return current_user

functions/sessionkey.py

@@ -0,0 +1,17 @@
+from datetime import timedelta, datetime
+from typing import Optional
+from jose import jwt
+
+import env as e
+import functions.hashing as hash
+
+
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(weeks=4)
+    to_encode.update({"exp": expire})
+    encode_jwt = jwt.encode(to_encode, e.SECRET_KEY, algorithm=hash.ALOGRITHM)
+    return encode_jwt

functions/user.py

@@ -1,3 +1,5 @@
+from fastapi import Depends
+
 from assets.database import openDBConnection
 from models.sessionkey import SessionKey
 from models.user import User
@@ -11,15 +13,74 @@ def get_user(ID=None, EMAIL=None, SESSION_KEY=None):
         cursor.execute("SELECT * FROM USERDATA WHERE ID = %s", (ID,))
         return fetch_data(cursor)
     if EMAIL is not None:
-        cursor.execute("SELECT * FROM USERDATA WHERE ID = %s", (EMAIL,))
+        cursor.execute("SELECT * FROM USERDATA WHERE EMAIL = %s", (EMAIL,))
         return fetch_data(cursor)
     if SESSION_KEY is not None:
-        cursor.execute("SELECT * FROM USERDATA WHERE ID = %s", (SESSION_KEY,))
+        cursor.execute("SELECT * FROM USERDATA WHERE SESSION_KEY = %s", (SESSION_KEY,))
         return fetch_data(cursor)
     else:
         raise ValueError('USER not Valid')
 
 
+def push_data(u: User):
+    db = openDBConnection()
+    cursor = db.cursor()
+
+    SQL = """
+    UPDATE USERDATA SET 
+    EMAIL = %s,
+    PASSWORD_HASH = %s,
+    NAME = %s,
+    LASTNAME = %s,
+    CLASS = %s,
+    PERMISSION_LEVEL = %s,
+    LAST_IP = %s,
+    ACTIVE = %s
+    WHERE ID = %s
+    """
+    PARAM = (
+        u.EMAIL,
+        u.PASSWORD_HASH,
+        u.NAME,
+        u.LASTNAME,
+        u.CLASS,
+        u.PERMISSION_LEVEL,
+        str(u.LAST_IP),
+        u.ACTIVE,
+        u.ID
+    )
+
+    cursor.execute(SQL, PARAM)
+    db.commit()
+    cursor.close()
+    db.close()
+
+
+def create_user(u: User):
+    db = openDBConnection()
+    cursor = db.cursor()
+
+    SQL = """
+    INSERT INTO USERDATA (EMAIL, PASSWORD_HASH, NAME, LASTNAME, CLASS, PERMISSION_LEVEL, LAST_IP, ACTIVE) 
+    VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
+    """
+    PARAM = (
+        u.EMAIL,
+        u.PASSWORD_HASH,
+        u.NAME,
+        u.LASTNAME,
+        u.CLASS,
+        u.PERMISSION_LEVEL,
+        str(u.LAST_IP),
+        u.ACTIVE
+    )
+
+    cursor.execute(SQL, PARAM)
+    db.commit()
+    cursor.close()
+    db.close()
+
+
 def fetch_data(cursor):
     data = cursor.fetchone()
     return User(
@@ -32,5 +93,19 @@ def fetch_data(cursor):
         PERMISSION_LEVEL=data[6],
         LAST_IP=data[7],
         ACTIVE=data[8],
-        SESSION_KEY=SessionKey(key=data[9])
     )
+
+
+def is_teacher(EMAIL: str):
+    db = openDBConnection()
+    cursor = db.cursor()
+
+    cursor.execute("SELECT EMAIL FROM GLOBAL_TEACHERS WHERE EMAIL = %s", (EMAIL,))
+
+    data: str = cursor.fetchone()
+
+    if data is not None:
+        if data[0].lower() == EMAIL.lower():
+            return True
+        return False
+    return False

main.py

@@ -1,13 +1,111 @@
-from fastapi import FastAPI
+from datetime import timedelta
+from typing import Optional
 
-import functions
-import models
-import assets
+from fastapi import FastAPI, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordRequestForm
+from pydantic import ValidationError
+from starlette.responses import JSONResponse
+
+from functions.hashing import get_current_active_user, authenticate_user, get_password_hash
+from functions.sessionkey import create_access_token
+from functions.user import create_user, get_user, is_teacher
+from models.token import Token
+from models.user import User
+from models.apikey import ApiKey
 
 app = FastAPI()
 
 
+# Post
+
+
+@app.post("/token", response_model=Token)
+async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
+    account = authenticate_user(form_data.username, form_data.password)
+    if not account:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access_token_expires = timedelta(weeks=4)
+    access_token = create_access_token(
+        data={"sub": account.EMAIL}, expires_delta=access_token_expires
+    )
+    return {"access_token": access_token, "token_type": "bearer"}
+
+
+# User get Operrations
+
+
+@app.get("/users/me", response_model=User, description="shows the current active logged in user")
+async def read_users_me(current_user: User = Depends(get_current_active_user)):
+    return current_user
+
+
+@app.get("/users/{id}", response_model=User)
+async def read_user_by_id(id: int, current_user: User = Depends(get_current_active_user)):
+    if current_user.PERMISSION_LEVEL >= 3:
+        return get_user(ID=id)
+    else:
+        return JSONResponse(status_code=status.HTTP_403_FORBIDDEN,
+                            content="Not sufficent Permissions to view other users")
+
+
+# Put
+
+@app.put("/admin/create/user")
+async def admin_create_user(user: User, apikey: Optional[ApiKey] = None,
+                            current_user: User = Depends(get_current_active_user)):
+    if current_user.PERMISSION_LEVEL >= 3:
+        user.PASSWORD_HASH = get_password_hash(user.PASSWORD_HASH)
+        create_user(user)
+        return JSONResponse(status_code=status.HTTP_201_CREATED, content="User was successfully created")
+    else:
+        return JSONResponse(
+            status_code=status.HTTP_403_FORBIDDEN,
+            content="Not sufficent Permissions to create other users"
+        )
+
+
+@app.put("/create/user")
+async def form_create_user(api_key: str, u_email: str, u_password: str, u_class: str):
+    try:
+        ApiKey(APIKEY=api_key)
+    except ValidationError as e:
+        return JSONResponse(
+            status_code=status.HTTP_403_FORBIDDEN,
+            content=e.errors()
+        )
+
+    PERMISSION_LEVEL = 0
+    if is_teacher(u_email):
+        PERMISSION_LEVEL = 1
 
+    NAME = u_email.split(".")[0].capitalize()
+    LASTNAME = u_email.split(".")[1].split("@")[0].capitalize()
 
+    if LASTNAME[:-2].isdigit():
+        LASTNAME = LASTNAME[:-2]
 
 
+    try:
+        account = User(
+            EMAIL=u_email,
+            PASSWORD_HASH=get_password_hash(u_password),
+            NAME=NAME,
+            LASTNAME=LASTNAME,
+            CLASS=u_class,
+            PERMISSION_LEVEL=PERMISSION_LEVEL,
+            ACTIVE=False
+        )
+    except ValidationError as e:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=e.errors()
+        )
+    create_user(account)
+    return JSONResponse(
+        status_code=status.HTTP_200_OK,
+        content="User succesfully created"
+    )

models/sessionkey.py

@@ -4,19 +4,21 @@
 
 
 class SessionKey(BaseModel):
-    key: str
+    key: Optional[str] = None
     valid: Optional[bool] = False
 
-    @root_validator
-    def check_session_key(cls, values):
-        db = openDBConnection()
-        cursor = db.cursor()
+    # @root_validator
+    # def check_session_key(cls, values):
+    #     db = openDBConnection()
+    #     cursor = db.cursor()
+    #
+    #     cursor.execute("SELECT SESSION_KEY FROM USERDATA WHERE SESSION_KEY = %s", (values['key'],))
+    #     data = cursor.fetchone()
+    #
+    #     if data[0] is not None:
+    #         values['valid'] = True
+    #         return values
+    #
+    #     raise ValueError('SessionKey is not valid')
 
-        cursor.execute("SELECT SESSION_KEY FROM USERDATA WHERE SESSION_KEY = %s", (values['key'],))
-        data = cursor.fetchone()
 
-        if data[0] is not None:
-            values['valid'] = True
-            return values
-
-        raise ValueError('SessionKey is not valid')

models/token.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str

models/tokendata.py

@@ -0,0 +1,7 @@
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class TokenData(BaseModel):
+    EMAIL: Optional[str] = None