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 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'] if ipaddress.ip_address(ip) in ipaddress.ip_network('192.168.0.0/16'): 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}"}) else: return templates.TemplateResponse(request=request, name='unknown_user.html',context={"IP":ip}) @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 ipaddress.ip_address(ip) in ipaddress.ip_network('192.168.0.0/16'): active_users = get_active_users() return templates.TemplateResponse(request=request, name='user.html',context={"username":username, "active_users": active_users}) else: return templates.TemplateResponse(request=request, name='unknown_user.html',context={"IP":ip}) 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 # @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)