fix: corrige path traversal, fuites d'erreurs et dépréciations

- Empêche le path traversal du catch-all SPA (vérif is_relative_to dist)
- Migre on_event("startup") vers le handler lifespan de FastAPI
- Masque les exceptions internes des erreurs 500 (log au lieu de str(e))
- Remplace datetime.utcnow déprécié par un helper UTC timezone-aware
- Simplifie except (ValueError, Exception) redondant

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-11 10:01:48 +02:00
parent bd8b9a044c
commit c15b65f0d9
5 changed files with 44 additions and 28 deletions

View File

@@ -1,5 +1,6 @@
"""Application FastAPI pour l'extraction de comptes rendus de gérance."""
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI
@@ -19,6 +20,13 @@ from .routes import (
config_router,
)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Initialize database on application startup."""
init_db()
yield
app = FastAPI(
title="Plesna Gérance API",
description="API pour extraire les informations structurées des PDFs de comptes rendus de gérance Oralia/ICS.",
@@ -26,16 +34,10 @@ app = FastAPI(
docs_url="/api/docs",
redoc_url="/api/redoc",
openapi_url="/api/openapi.json",
lifespan=lifespan,
)
# Initialize database on startup
@app.on_event("startup")
async def startup_event():
"""Initialize database on application startup."""
init_db()
# Include routers
app.include_router(extraction_router)
app.include_router(documents_router)
@@ -74,12 +76,14 @@ if FRONTEND_DIST.exists():
app.mount("/assets", StaticFiles(directory=FRONTEND_DIST / "assets"), name="assets")
# Catch-all route for SPA - must be last
_dist_root = FRONTEND_DIST.resolve()
@app.get("/{full_path:path}", include_in_schema=False)
async def serve_spa(full_path: str):
"""Serve the SPA for all non-API routes."""
# If requesting a file that exists, serve it
file_path = FRONTEND_DIST / full_path
if file_path.is_file():
# If requesting a file that exists (and stays within dist), serve it
file_path = (FRONTEND_DIST / full_path).resolve()
if file_path.is_file() and file_path.is_relative_to(_dist_root):
return FileResponse(file_path)
# Otherwise serve index.html for SPA routing
return FileResponse(FRONTEND_DIST / "index.html")
return FileResponse(_dist_root / "index.html")

View File

@@ -1,6 +1,7 @@
"""Documents routes - CRUD operations for documents."""
import json
import logging
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, File, UploadFile, Form
@@ -16,6 +17,8 @@ from ..schemas import SaveRequest, SaveResponse, DocumentSummary
router = APIRouter(prefix="/api", tags=["documents"])
logger = logging.getLogger(__name__)
@router.post("/save", response_model=SaveResponse)
async def save_document(
@@ -60,9 +63,10 @@ async def save_document(
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
except Exception:
logger.exception("Erreur lors de la sauvegarde du document")
raise HTTPException(
status_code=500, detail=f"Erreur lors de la sauvegarde: {str(e)}"
status_code=500, detail="Erreur lors de la sauvegarde du document."
)
@@ -134,9 +138,10 @@ async def save_document_with_pdf(
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
except Exception:
logger.exception("Erreur lors de la sauvegarde du document avec PDF")
raise HTTPException(
status_code=500, detail=f"Erreur lors de la sauvegarde: {str(e)}"
status_code=500, detail="Erreur lors de la sauvegarde du document."
)

View File

@@ -76,9 +76,11 @@ async def ask_ia(request: AskRequest) -> AskResponse:
status_code=503,
detail=f"Erreur Ollama : {e.response.status_code}",
)
except Exception as e:
logger.error(f"IA ask error: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
except Exception:
logger.exception("IA ask error")
raise HTTPException(
status_code=500, detail="Erreur interne lors du traitement de la question."
)
@router.get("/health", response_model=HealthResponse)

View File

@@ -1,6 +1,6 @@
"""SQLAlchemy models for plesna-gerance database."""
from datetime import date, datetime
from datetime import date, datetime, timezone
from typing import Optional
from sqlalchemy import (
@@ -18,6 +18,11 @@ from sqlalchemy import (
from sqlalchemy.orm import DeclarativeBase, relationship
def _utcnow() -> datetime:
"""Horodatage UTC timezone-aware (remplace datetime.utcnow déprécié)."""
return datetime.now(timezone.utc)
class Base(DeclarativeBase):
"""Base class for all models."""
@@ -31,7 +36,7 @@ class Tag(Base):
id = Column(Integer, primary_key=True, autoincrement=True)
nom = Column(String(100), unique=True, nullable=False, index=True)
created_at = Column(DateTime, default=datetime.utcnow)
created_at = Column(DateTime, default=_utcnow)
# Relations
depenses = relationship("Depense", back_populates="tag")
@@ -48,7 +53,7 @@ class Setting(Base):
id = Column(Integer, primary_key=True, autoincrement=True)
key = Column(String(100), unique=True, nullable=False, index=True)
value = Column(Text, nullable=False)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
updated_at = Column(DateTime, default=_utcnow, onupdate=_utcnow)
def __repr__(self) -> str:
return f"<Setting(key={self.key}, value={self.value})>"
@@ -64,7 +69,7 @@ class Immeuble(Base):
adresse = Column(String(255), nullable=True)
ville = Column(String(100), nullable=True)
code_postal = Column(String(10), nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
created_at = Column(DateTime, default=_utcnow)
# Relations
lots = relationship("Lot", back_populates="immeuble", cascade="all, delete-orphan")
@@ -84,7 +89,7 @@ class Lot(Base):
immeuble_id = Column(Integer, ForeignKey("immeubles.id"), nullable=False)
numero = Column(String(10), nullable=False)
type = Column(String(50), nullable=True) # "Loc. Commercial", "Appartement", etc.
created_at = Column(DateTime, default=datetime.utcnow)
created_at = Column(DateTime, default=_utcnow)
# Contrainte unique: un numéro de lot par immeuble
__table_args__ = (
@@ -114,7 +119,7 @@ class Locataire(Base):
nom = Column(String(255), nullable=False)
date_debut = Column(Date, nullable=True) # Date d'entrée dans le lot
date_fin = Column(Date, nullable=True) # Date de sortie (NULL si actif)
created_at = Column(DateTime, default=datetime.utcnow)
created_at = Column(DateTime, default=_utcnow)
# Contrainte unique: un locataire par lot et période
__table_args__ = (
@@ -156,7 +161,7 @@ class Document(Base):
solde_type = Column(String(20), nullable=True) # "crediteur" ou "debiteur"
solde_date_arrete = Column(Date, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
created_at = Column(DateTime, default=_utcnow)
# Chemins vers les fichiers stockés (relatifs à PLESNA_STORAGE_PATH)
pdf_path = Column(
@@ -211,7 +216,7 @@ class Revenu(Base):
regles = Column(Float, default=0.0) # Montant réglé
impayes = Column(Float, default=0.0)
created_at = Column(DateTime, default=datetime.utcnow)
created_at = Column(DateTime, default=_utcnow)
__table_args__ = (
Index("ix_revenu_document", "document_id"),
@@ -256,7 +261,7 @@ class Depense(Base):
locatif = Column(Float, default=0.0) # Part locative
deductible = Column(Float, default=0.0) # Part déductible
created_at = Column(DateTime, default=datetime.utcnow)
created_at = Column(DateTime, default=_utcnow)
__table_args__ = (
Index("ix_depense_document", "document_id"),

View File

@@ -148,7 +148,7 @@ def _execute_tool(name: str, arguments: dict) -> str:
try:
result = execute_readonly_sql(query)
return json.dumps(result, ensure_ascii=False, default=str)
except (ValueError, Exception) as e:
except Exception as e:
return json.dumps({"error": str(e)}, ensure_ascii=False)
else:
return json.dumps({"error": f"Outil inconnu : {name}"})