PythonFastServer/main.py

from fastapi import FastAPI, Form, File, Request
from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse
from jinja2 import Template
import os
from typing import Annotated
from pydantic import BaseModel
import tempfile
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
import ipaddress
import psycopg2
from dateutil.parser import parse
from datetime import datetime
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)

temp_dir = r"/home/td/WebTemp"
tempfile.tempdir = temp_dir
con = psycopg2.connect("host=127.0.0.1 dbname=test user=alfred password='Alf red'")


app = FastAPI(title="user manager", description="User session management.", version="0.0.0.1",
               docs_url="/server/docs", redoc_url="/server/redoc", openapi_url="/server/openapi.json")

app.mount("/server/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")

@app.get("/server", response_class=HTMLResponse)
async def get_server(request: Request):
    ip = request.headers['X-Real-IP']
    c = request.cookies.get("failed_reason", None)
    if c is None:
        return templates.TemplateResponse(request=request, name='index.html',context={"status":""})
    else:
        return templates.TemplateResponse(request=request, name='index.html',context={"status":f"Failed to login: {c}"})


@app.get("/server/user", response_class=HTMLResponse)
async def get_user_server(request: Request):
    ip = request.headers['X-Real-IP']
    username = request.cookies.get('username', None)
    session_key = request.cookies.get('session_key', None)
    if username is None or len(username) == 0:
        return RedirectResponse("/server", status_code=303)
    if session_key is None or len(session_key) == 0:
        return RedirectResponse("/server", status_code=303)
    if not validate_session_key(username, session_key):
        r = RedirectResponse("/server", status_code=303)
        r.set_cookie("failed_reason", "Could not validate user session")
        return r
    active_users = get_active_users()
    return templates.TemplateResponse(request=request, name='user.html',context={"username":username, "active_users": active_users})


def get_active_users() -> list[list]:
    active_users = []
    cur = con.cursor()
    cur.execute("Select * from API_USERS where session_status = 'ACTIVE'")
    results = cur.fetchall()
    print(f"Results -> {len(results)}")
    for i in results:
        j = [i[0], i[3]]
        active_users.append(j)
    print("Active users: ", active_users)
    return active_users


def validate_session_key(username, session_key) -> bool:
    cur = con.cursor()
    cur.execute(f"Select username, session_key, session_status, expiry_time from API_USERS where session_status='ACTIVE' and username='{username}'")
    results = cur.fetchall()
    if len(results) <= 0:
        return False
    (q_username, q_session_key, q_session_status, q_expiry_time) = results[0]
    print("Len ->", len(results))
    if q_username == username and q_session_key == session_key:
        now_time = datetime.now()
        q_expiry_time = q_expiry_time
        print("Expiry time", q_expiry_time, now_time)
        if q_expiry_time < now_time:
            return False
        return True
    else:
        return False
    


# @app.post("/api/users/create", response_class=RedirectResponse)
# async def root(username: Annotated[str, Form()], password: Annotated[str, Form()],
#                new_user: Annotated[str, Form()], new_password: Annotated[str, Form()]):
#      global internal_status
#      val = helper.validate_credentials(username, password)
#      print("Validation result: ", val)
#      if val == "valid":
#         print("Valid creds")
#         helper.status_type = 'success'
#         helper.internal_status = "Created New User"
#         helper.create_new_user(new_user, new_password)
#         return RedirectResponse(url=f"/api/users/", status_code=303)
#      elif val == "invalid":
#          print("Invalid creds")
#          helper.status_type = 'danger'
#          helper.internal_status = "Invalid Credentials"
#          return RedirectResponse(url=f"/api/users/", status_code=303)
#      else:
#         print("Validation issue: ", val)
#         helper.status_type = 'danger'
#         helper.internal_status = val
#         return RedirectResponse(url="/api/users/", status_code=303)
     
# @app.post("/api/users/login")
# async def root(user: User, request: Request):
#     ip = request.headers['X-Real-IP']
#     if ipaddress.ip_address(ip) not in ipaddress.ip_network('192.168.0.0/16'):
#         return {"message": "unknown user"}
#     val = helper.validate_credentials(user.username, user.password)
#     if val == 'valid':
#         print("Valid creds generating session key")
#         (session_key, expiry_date) = helper.create_new_session_key(user.username)
#         return {"message": "credentials validated", "session_key": session_key, "expiry_date": expiry_date}
#     elif val == 'invalid':
#         print("Invalid creds.")
#         return {"message": "invalid credentials"}
#     elif val == 'session_active':
#         print("session is active.")
#         return {"message": "active session"}
    

# @app.post("/api/users/logout")
# async def root(user: User, request: Request):
#     ip = request.headers['X-Real-IP']
#     if ipaddress.ip_address(ip) not in ipaddress.ip_network('192.168.0.0/16'):
#         return {"message": "unknown user"}
#     val = helper.validate_credentials(user.username, user.password)
#     if val == 'valid':
#         print(f"User {user.username} is not active.")
#         return {"message", "user not active"}
#     elif val == 'session_active':
#         print(f"Logging out user: {user.username}, bye!")
#         ret = helper.logout_user(user.username)
#         return {"message": f"logout status: {str(ret)}"}
#     else:
#         print(f"invalid credentials.")
#         return {"message": "invalid credentials"}





# @app.exception_handler(RequestValidationError)
# async def validation_exception_handler(request, exc):
#     print(exc)
#     return PlainTextResponse(str(exc), status_code=400)


# @app.exception_handler(StarletteHTTPException)
# async def custom_http_exception_handler(request, exc):
#     print(f"OMG! An HTTP error!: {repr(exc)}")
#     return await http_exception_handler(request, exc)


# @app.exception_handler(RequestValidationError)
# async def validation_exception_handler(request, exc):
#     print(f"OMG! The client sent invalid data!: {exc}")
#     return await request_validation_exception_handler(request, exc)