feat: implement session_jwt and enhance auth

This commit is contained in:
Siddhant Rai
2025-03-17 11:51:30 +05:30
parent 7fd377bdbe
commit 4406426515
9 changed files with 148 additions and 59 deletions

View File

@@ -1,10 +1,12 @@
import os
import platform
import uuid
import dotenv
from flask import Flask, jsonify, redirect, request
from jose import jwt
from application.auth import get_or_create_user_id, handle_auth
from application.auth import handle_auth
from application.core.logging_config import setup_logging
@@ -38,10 +40,22 @@ app.config.update(
celery.config_from_object("application.celeryconfig")
api.init_app(app)
if settings.AUTH_TYPE in ("simple_jwt", "session_jwt") and not settings.JWT_SECRET_KEY:
key_file = ".jwt_secret_key"
try:
with open(key_file, "r") as f:
settings.JWT_SECRET_KEY = f.read().strip()
except FileNotFoundError:
new_key = os.urandom(32).hex()
with open(key_file, "w") as f:
f.write(new_key)
settings.JWT_SECRET_KEY = new_key
except Exception as e:
raise RuntimeError(f"Failed to setup JWT_SECRET_KEY: {e}")
SIMPLE_JWT_TOKEN = None
if settings.AUTH_TYPE == "simple_jwt":
user_id = get_or_create_user_id()
payload = {"sub": user_id}
payload = {"sub": "local"}
SIMPLE_JWT_TOKEN = jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256")
print(f"Generated Simple JWT Token: {SIMPLE_JWT_TOKEN}")
@@ -54,13 +68,33 @@ def home():
return "Welcome to DocsGPT Backend!"
@app.route("/api/config")
def get_config():
response = {
"auth_type": settings.AUTH_TYPE,
"requires_auth": settings.AUTH_TYPE in ["simple_jwt", "session_jwt"],
}
return jsonify(response)
@app.route("/api/generate_token")
def generate_token():
if settings.AUTH_TYPE == "session_jwt":
new_user_id = str(uuid.uuid4())
token = jwt.encode(
{"sub": new_user_id}, settings.JWT_SECRET_KEY, algorithm="HS256"
)
return jsonify({"token": token})
return jsonify({"error": "Token generation not allowed in current auth mode"}), 400
@app.before_request
def authenticate_request():
if request.method == "OPTIONS":
return "", 200
decoded_token = handle_auth(request)
if "message" in decoded_token:
if not decoded_token:
request.decoded_token = None
elif "error" in decoded_token:
return jsonify(decoded_token), 401

View File

@@ -1,15 +1,13 @@
import uuid
from jose import jwt
from application.core.settings import settings
def handle_auth(request, data={}):
if settings.AUTH_TYPE == "simple_jwt":
if settings.AUTH_TYPE in ["simple_jwt", "session_jwt"]:
jwt_token = request.headers.get("Authorization")
if not jwt_token:
return {"message": "Missing Authorization header"}
return None
jwt_token = jwt_token.replace("Bearer ", "")
@@ -22,18 +20,9 @@ def handle_auth(request, data={}):
)
return decoded_token
except Exception as e:
return {"message": f"Authentication error: {str(e)}"}
return {
"message": f"Authentication error: {str(e)}",
"error": "invalid_token",
}
else:
return {"sub": "local"}
def get_or_create_user_id():
try:
with open(settings.USER_ID_FILE, "r") as f:
user_id = f.read().strip()
return user_id
except FileNotFoundError:
user_id = str(uuid.uuid4())
with open(settings.USER_ID_FILE, "w") as f:
f.write(user_id)
return user_id

View File

@@ -100,7 +100,6 @@ class Settings(BaseSettings):
FLASK_DEBUG_MODE: bool = False
JWT_SECRET_KEY: str = ""
USER_ID_FILE: str = os.path.join(current_dir, "user_id.txt")
path = Path(__file__).parent.parent.absolute()