Added user authentication; still some todo\'s

backend/app.py

@@ -1,7 +1,7 @@
 import uvicorn
 from fastapi import FastAPI
 
-from routes.session import session_router
+from routes.authentication import session_router
 from routes.teachers import teachers_router
 
 app = FastAPI()

backend/application.properties

@@ -0,0 +1,4 @@
+[session]
+service = https://localhost:8080/api/login
+cookie_domain = localhost
+max_cookie_age = 86400

backend/controllers/auth/session_controller.py → backend/controllers/auth/authentication_controller.py

@@ -1,15 +1,14 @@
 import httpx
 from defusedxml.ElementTree import fromstring
 
-# test url: https://login.ugent.be/login?service=https://localhost:8080/session
-# TODO: get information out of a properties file
-SERVICE = "https://localhost:8080/session"
-DOMAIN = "localhost"
-MAX_AGE = 24 * 60 * 60
+from controllers.properties.Properties import Properties
 
+props: Properties = Properties()
 
-def get_user_information(ticket: str) -> dict | None:
-    user_information = httpx.get(f"https://login.ugent.be/serviceValidate?service={SERVICE}&ticket={ticket}"
+
+def authenticate_user(ticket: str) -> dict | None:
+    service = props.get("session", "service")
+    user_information = httpx.get(f"https://login.ugent.be/serviceValidate?service={service}&ticket={ticket}"
                                  , headers={"Accept": "application/json,text/html"},
                                  )
     user: dict | None = parse_cas_xml(user_information.text)
@@ -31,11 +30,6 @@ def parse_cas_xml(xml: str) -> dict | None:
         mail = attributes_xml.find(f"{namespace}mail").text
 
         user["name"] = f"{givenname} {surname}"
-        user["mail"] = mail
+        user["mail"] = mail.lower()
         return user
     return None
-
-
-# TODO: create a session_id for the given user, create the user if it doesn't exist already
-def login_user(user_information: dict) -> str:
-    return "TestValueCookie"

backend/controllers/auth/cookie_controller.py

@@ -0,0 +1,27 @@
+from fastapi import Request, Response
+
+from controllers.auth.encryption_controller import encrypt, generate_keys
+from controllers.properties.Properties import Properties
+
+props: Properties = Properties()
+
+
+def set_cookies(response: Response, key: str, value: str) -> Response:
+    value: str = encrypt(value)
+    max_age: int = int(props.get("session", "max_cookie_age"))
+    domain: str = props.get("session", "cookie_domain")
+    response.set_cookie(key=key,
+                        value=value,
+                        max_age=max_age,
+                        domain=domain,
+                        secure=False)
+    return response
+
+
+def get_cookie(request: Request, key: str) -> str:
+    return request.cookies.get(key)
+
+
+def delete_cookie(response: Response, cookie_tag: str) -> Response:
+    response.delete_cookie(cookie_tag)
+    return response

backend/controllers/auth/encryption_controller.py

@@ -0,0 +1,53 @@
+import base64
+
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import padding, rsa
+from cryptography.hazmat.primitives.asymmetric.padding import OAEP
+from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey, RSAPublicKey
+
+# TODO: Use database to manage keys instead of dict
+keys: dict = {}
+
+
+def generate_keys(user_id: str) -> RSAPublicKey:
+    if user_id in keys:
+        return keys[user_id].public_key()
+
+    private_key: RSAPrivateKey = rsa.generate_private_key(
+        public_exponent=65537,
+        key_size=2048,
+    )
+    keys[user_id] = private_key
+    return private_key.public_key()
+
+
+def encrypt(user_id: str) -> str:
+    key = generate_keys(user_id)
+    encryption: bytes = key.encrypt(
+        plaintext=user_id.encode(),
+        padding=OAEP(
+            mgf=padding.MGF1(algorithm=hashes.SHA256()),
+            algorithm=hashes.SHA256(),
+            label=None,
+        ),
+    )
+    return base64.b64encode(encryption).decode()
+
+
+def decrypt(user_id: str, encrypted_text: str) -> str | None:
+    if user_id in keys:
+        private_key: RSAPrivateKey = keys[user_id]
+        encryption: bytes = base64.b64decode(encrypted_text.encode())
+        return private_key.decrypt(
+            encryption,
+            padding=OAEP(
+                mgf=padding.MGF1(algorithm=hashes.SHA256()),
+                algorithm=hashes.SHA256(),
+                label=None,
+            ),
+        ).decode()
+    return None
+
+
+def delete_key(user_id: str) -> None:
+    keys.pop(user_id)

backend/controllers/auth/login_controller.py

@@ -0,0 +1,16 @@
+from fastapi import Request
+
+from controllers.auth.cookie_controller import get_cookie
+from controllers.auth.encryption_controller import decrypt
+from controllers.properties.Properties import Properties
+
+# test url: https://login.ugent.be/login?service=https://localhost:8080/api/login
+props: Properties = Properties()
+
+
+def verify_session(request: Request, user_id: str) -> bool:
+    session: str = get_cookie(request, "session_id")
+    if session:
+        decryption = decrypt(user_id=user_id, encrypted_text=session)
+        return user_id == decryption if decryption else False
+    return False

backend/controllers/properties/Properties.py

@@ -0,0 +1,15 @@
+import configparser
+
+
+class Properties:
+    def __init__(self) -> None:
+        config = configparser.ConfigParser()
+        config.read("application.properties")
+        self.properties: dict = {}
+        for section in config.sections():
+            self.properties[section] = {}
+            for key, val in config[section].items():
+                self.properties[section][key] = val
+
+    def get(self, section: str, key: str) -> str:
+        return self.properties[section][key]

backend/requirements.txt

@@ -20,4 +20,5 @@ starlette==0.36.3
 typing_extensions==4.10.0
 uvicorn==0.27.1
 httpx==0.27.0
-defusedxml~=0.7.1
+defusedxml~=0.7.1
+cryptography~=42.0.5

backend/routes/authentication.py

@@ -0,0 +1,56 @@
+from fastapi import APIRouter, Request, Response
+from fastapi.responses import JSONResponse
+
+from controllers.auth.authentication_controller import authenticate_user
+from controllers.auth.cookie_controller import delete_cookie, set_cookies
+from controllers.auth.encryption_controller import delete_key
+from controllers.auth.login_controller import verify_session
+
+session_router = APIRouter()
+
+
+@session_router.get("/api/login")
+def login(ticket: str) -> JSONResponse:
+    """
+    This function start a session for the user.
+    For authentication, it uses the given ticket and the UGent CAS server (https://login.ugent.be).
+
+    :param ticket: A UGent CAS ticket that will be used for the authentication
+    :return:
+        - Valid Ticket: A JSONResponse with a user object; a cookie will be set with a session_id
+        - Invalid Ticket: A JSONResponse with status_code 401 and an error message
+    """
+    user: dict = authenticate_user(ticket)  # This should be a user object
+    if user:
+        response: JSONResponse = JSONResponse(content=user)
+        # TODO: Change mail to user id
+        print("here")
+        response = set_cookies(response, "session_id", user["mail"])
+        return response
+    return JSONResponse(status_code=401, content="Invalid Ticket")
+
+
+@session_router.get("/api/logout/{user_id}")
+def logout(user_id: str) -> Response:
+    """
+    This function will log a user out, by removing the session from storage
+
+    :param user_id: An identifier of the user that needs to be logged out
+    :return: A confirmation that the logout was successful, and it tells the browser to remove the cookie.
+    """
+    response: Response = Response(content="You've been successfully logged out")
+
+    delete_cookie(response, "session_id")
+    delete_key(user_id)
+    return response
+
+
+@session_router.get("/api/session/verify/{user_id}")
+def verify(request: Request, user_id: str) -> bool:
+    """
+    A test route to check if the user has a valid session
+    :param request: Http Request filled in by fastapi
+    :param user_id: identifier for the user
+    :return: boolean that says if the user is logged in
+    """
+    return verify_session(request, user_id)