Spaces:
Runtime error
Runtime error
File size: 3,084 Bytes
aebb938 03c4b86 aebb938 03c4b86 aebb938 03c4b86 aebb938 03c4b86 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | import os
from fastapi import FastAPI, Request, Form, Depends, HTTPException, status
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel
from typing import Optional
from datetime import datetime, timedelta
from jose import JWTError, jwt
app = FastAPI()
# Configuration for dummy Auth
SECRET_KEY = "dummy_secret_key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
# Dummy user database: username -> {password, role}
USER_DB = {
"auditor1": {"password": "password123", "role": "auditor"},
"reviewer1": {"password": "admin123", "role": "reviewer"}
}
templates = Jinja2Templates(directory=".")
def create_access_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def get_current_user(request: Request):
token = request.cookies.get("access_token")
if not token:
return None
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
return None
return {"username": username, "role": payload.get("role")}
except JWTError:
return None
@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
user = get_current_user(request)
if not user:
return templates.TemplateResponse("index.html", {"request": request, "user": None})
return templates.TemplateResponse("index.html", {"request": request, "user": user})
@app.post("/login")
async def login(username: str = Form(...), password: str = Form(...)):
user = USER_DB.get(username)
if user and user["password"] == password:
token = create_access_token({"sub": username, "role": user["role"]})
response = RedirectResponse(url="/", status_code=status.HTTP_302_FOUND)
response.set_cookie(key="access_token", value=token, httponly=True)
return response
return HTMLResponse("Invalid credentials", status_code=401)
@app.get("/logout")
async def logout():
response = RedirectResponse(url="/")
response.delete_cookie("access_token")
return response
@app.post("/submit-finding")
async def submit_finding(request: Request, finding: str = Form(...)):
user = get_current_user(request)
if not user or user["role"] != "auditor":
raise HTTPException(status_code=403, detail="Only auditors can submit findings")
# Logic to save finding would go here
return {"status": "success", "message": "Finding submitted for review"}
@app.post("/review-finding")
async def review_finding(request: Request, decision: str = Form(...)):
user = get_current_user(request)
if not user or user["role"] != "reviewer":
raise HTTPException(status_code=403, detail="Only reviewers can approve findings")
return {"status": "success", "message": f"Finding {decision}"}
|