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)