Public Access
131 lines
4.8 KiB
Python
131 lines
4.8 KiB
Python
import os
|
|
from dotenv import load_dotenv
|
|
load_dotenv() # Force la lecture du fichier .env en local
|
|
import uuid
|
|
import json
|
|
from fasthtml.common import *
|
|
from starlette.staticfiles import StaticFiles
|
|
from starlette.responses import FileResponse, Response, HTMLResponse, RedirectResponse
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
|
|
from modules.theme import get_theme_hdrs
|
|
from modules.database import nettoyer_vieilles_discussions, HISTORY_FILE, load_history, save_history
|
|
from modules.ui_components import render_main_page, render_sidebar, render_login_page
|
|
from modules.chat_handler import handle_chat_request
|
|
from modules.security import verifier_et_bloquer_vpn, journaliser_ip
|
|
from modules.auth_manager import IDENTIFIANTS_ADMIN, verifier_mot_de_passe
|
|
from modules.logger import log_event
|
|
from modules.billing_manager import charger_soldes, sauvegarder_soldes
|
|
|
|
# Lancement du module d'auto-mise à jour en arrière-plan
|
|
import modules.model_updater
|
|
|
|
# --- MIDDLEWARE DE SÉCURITÉ ---
|
|
class SecurityMiddleware(BaseHTTPMiddleware):
|
|
async def dispatch(self, request, call_next):
|
|
ip = request.headers.get("X-Forwarded-For", request.client.host).split(",")[0].strip()
|
|
utilisateur = request.cookies.get("aethas_user", "visiteur")
|
|
|
|
if not request.url.path.startswith("/Logo") and not request.url.path.startswith("/login"):
|
|
if verifier_et_bloquer_vpn(ip, utilisateur):
|
|
return HTMLResponse("<h1 style='color:red; text-align:center;'>Accès refusé</h1><p style='text-align:center;'>VPN ou Proxy strictement interdit.</p>", status_code=403)
|
|
|
|
# Exclusion stricte du spam de logs (Silence pour /ping et /favicon)
|
|
if request.url.path not in ["/ping", "/favicon.ico"]:
|
|
journaliser_ip(ip, f"Accès à {request.url.path}")
|
|
|
|
return await call_next(request)
|
|
|
|
nettoyer_vieilles_discussions()
|
|
|
|
app, rt = fast_app(
|
|
hdrs=get_theme_hdrs(),
|
|
secret_key="super_secret_aethas",
|
|
middleware=[Middleware(SecurityMiddleware)]
|
|
)
|
|
|
|
app.mount("/Logo", StaticFiles(directory="Logo"), name="Logo")
|
|
|
|
# --- ROUTES D'AUTHENTIFICATION ---
|
|
@rt('/login')
|
|
def get():
|
|
return render_login_page()
|
|
|
|
@rt('/login')
|
|
async def post(request, session):
|
|
form = await request.form()
|
|
if form.get("email") == IDENTIFIANTS_ADMIN["email"] and verifier_mot_de_passe(form.get("password"), IDENTIFIANTS_ADMIN["mot_de_passe_hash"]):
|
|
session['user'] = "xavier"
|
|
resp = RedirectResponse('/', status_code=303)
|
|
resp.set_cookie("aethas_user", "xavier", max_age=86400)
|
|
return resp
|
|
return render_login_page(error="Accès refusé. Identifiants incorrects.")
|
|
|
|
@rt('/logout')
|
|
def get(session):
|
|
session.clear()
|
|
resp = RedirectResponse('/login', status_code=303)
|
|
resp.delete_cookie("aethas_user")
|
|
return resp
|
|
|
|
# --- ROUTES PROTÉGÉES ---
|
|
@rt('/')
|
|
def get(session, request):
|
|
utilisateur = session.get('user') or request.cookies.get('aethas_user')
|
|
if not utilisateur:
|
|
return RedirectResponse('/login', status_code=303)
|
|
|
|
session['user'] = utilisateur
|
|
if 'current_chat_id' not in session or not session['current_chat_id']:
|
|
session['current_chat_id'] = str(uuid.uuid4())
|
|
return render_main_page(session['current_chat_id'], utilisateur_nom=session['user'])
|
|
|
|
@rt('/chat')
|
|
async def post(request, session):
|
|
utilisateur = session.get('user') or request.cookies.get('aethas_user')
|
|
if not utilisateur:
|
|
return HTMLResponse(
|
|
"<div style='color:red; background:#300; padding:15px; border-radius:10px; margin-top:10px;'>"
|
|
"<b>❌ Session expirée ou non autorisée.</b> Veuillez rafraîchir la page (F5) pour vous reconnecter."
|
|
"</div>",
|
|
status_code=401
|
|
)
|
|
session['user'] = utilisateur
|
|
return await handle_chat_request(request, session)
|
|
|
|
@rt('/favicon.ico')
|
|
def favicon_get(): return Response(status_code=204)
|
|
|
|
@rt('/ping')
|
|
def get(): return Response(status_code=204)
|
|
|
|
@rt('/new_chat')
|
|
def get(session):
|
|
if 'user' in session: session['current_chat_id'] = str(uuid.uuid4())
|
|
return RedirectResponse('/', status_code=303)
|
|
|
|
@rt('/load_chat/{cid}')
|
|
def get(cid: str, session):
|
|
if cid and cid != "null" and 'user' in session: session['current_chat_id'] = cid
|
|
return RedirectResponse('/', status_code=303)
|
|
|
|
@rt('/rename/{cid}')
|
|
def post(cid: str, req):
|
|
h = load_history()
|
|
if cid in h and (nom := req.headers.get("HX-Prompt")): h[cid]['title'] = nom; save_history(h)
|
|
return render_sidebar()
|
|
|
|
@rt('/pin/{cid}')
|
|
def post(cid: str):
|
|
h = load_history()
|
|
if cid in h: h[cid]['pinned'] = not h[cid].get('pinned', False); save_history(h)
|
|
return render_sidebar()
|
|
|
|
@rt('/delete/{cid}')
|
|
def post(cid: str):
|
|
h, cid_str = load_history(), str(cid)
|
|
if cid_str in h: del h[cid_str]; save_history(h)
|
|
return render_sidebar()
|
|
|
|
if __name__ == '__main__':
|
|
serve(port=5001) |