OpenMind v1 : cartes mentales libres, local-first et dockerisees

Premiere version fonctionnelle, demarrable avec `docker compose up -d --build`
sur http://localhost:8080.

Architecture
- Frontend React 18 / TypeScript strict / Vite / Tailwind, canvas @xyflow/react,
  etat Zustand, persistance IndexedDB via Dexie, auto-layout dagre.
- Backend FastAPI / SQLAlchemy 2 / Pydantic v2 / SQLite, qui sert aussi les
  fichiers statiques du frontend compile.
- Image Docker multi-stage, un seul service, bind mount ./data pour SQLite.

Local-first
Les cartes restent dans le navigateur. Le serveur ne recoit une carte que sur
demande explicite de partage.

Partage
Deux secrets distincts : share_token public en lecture seule dans l'URL, et
management_token conserve uniquement cote client, transmis en en-tete pour les
modifications et stocke hashe en base.

Fonctionnalites
Tableau de bord, editeur complet (undo/redo, copier/coller, raccourcis,
branches repliables, recherche), 9 themes, 4 dispositions automatiques,
panneau de proprietes, import/export .mmap, export PNG/SVG/PDF, impression,
page publique en lecture seule, PWA.

Qualite
25 tests pytest et 43 tests vitest au vert, ruff et ESLint sans erreur,
TypeScript strict sans erreur.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-08 22:19:18 +02:00
commit 0b2e5e326d
113 changed files with 14576 additions and 0 deletions

0
backend/app/__init__.py Normal file
View File

View File

@@ -0,0 +1,11 @@
"""Agregation des routeurs sous le prefixe /api."""
from fastapi import APIRouter
from app.api import health, shares
api_router = APIRouter(prefix="/api")
api_router.include_router(health.router)
api_router.include_router(shares.router)
__all__ = ["api_router"]

29
backend/app/api/deps.py Normal file
View File

@@ -0,0 +1,29 @@
"""Dependances communes aux routes."""
from __future__ import annotations
from typing import Annotated
from fastapi import Depends, Header
from sqlalchemy.orm import Session
from app.config import Settings, get_settings
from app.db.session import get_db
from app.services.errors import ShareForbiddenError
MANAGEMENT_TOKEN_HEADER = "X-Management-Token"
DbSession = Annotated[Session, Depends(get_db)]
AppSettings = Annotated[Settings, Depends(get_settings)]
def require_management_token(
x_management_token: Annotated[str | None, Header(alias=MANAGEMENT_TOKEN_HEADER)] = None,
) -> str:
"""Recupere le secret de gestion depuis l'en-tete (jamais depuis l'URL)."""
if not x_management_token:
raise ShareForbiddenError("Jeton de gestion manquant.")
return x_management_token
ManagementToken = Annotated[str, Depends(require_management_token)]

25
backend/app/api/health.py Normal file
View File

@@ -0,0 +1,25 @@
"""Endpoints de sante et de configuration publique."""
from __future__ import annotations
from fastapi import APIRouter
from app.api.deps import AppSettings
router = APIRouter(tags=["system"])
@router.get("/health", summary="Etat du service")
def health() -> dict[str, str]:
return {"status": "ok"}
@router.get("/config", summary="Configuration publique exposee au frontend")
def public_config(settings: AppSettings) -> dict[str, object]:
"""Aucun secret ici : uniquement ce dont l'interface a besoin."""
return {
"appName": settings.app_name,
"maxNodes": settings.max_nodes,
"maxMapSizeMb": settings.max_map_size_mb,
"sharingEnabled": True,
}

98
backend/app/api/shares.py Normal file
View File

@@ -0,0 +1,98 @@
"""API REST de partage de cartes."""
from __future__ import annotations
from fastapi import APIRouter, Response, status
from app.api.deps import DbSession, ManagementToken
from app.schemas.share import (
ShareCreatedOut,
ShareCreateIn,
SharePublicOut,
ShareUpdatedOut,
)
from app.services import shares as share_service
router = APIRouter(prefix="/shares", tags=["shares"])
def _share_path(share_token: str) -> str:
return f"/s/{share_token}"
@router.post(
"",
response_model=ShareCreatedOut,
status_code=status.HTTP_201_CREATED,
summary="Publier un snapshot de carte et obtenir un lien non repertorie",
)
def create_share(payload: ShareCreateIn, session: DbSession) -> ShareCreatedOut:
share, management_token = share_service.create_share(
session, title=payload.title, document=payload.map
)
return ShareCreatedOut(
share_token=share.share_token,
management_token=management_token,
share_path=_share_path(share.share_token),
title=share.title,
created_at=share.created_at,
updated_at=share.updated_at,
)
@router.get(
"/{share_token}",
response_model=SharePublicOut,
summary="Lire une carte partagee (public, lecture seule)",
)
def read_share(share_token: str, session: DbSession) -> SharePublicOut:
share = share_service.get_share(session, share_token)
return SharePublicOut(
share_token=share.share_token,
title=share.title,
map_version=share.map_version,
map=share_service.load_map_data(share),
created_at=share.created_at,
updated_at=share.updated_at,
)
@router.put(
"/{share_token}",
response_model=ShareUpdatedOut,
summary="Mettre a jour le snapshot partage (necessite le jeton de gestion)",
)
def update_share(
share_token: str,
payload: ShareCreateIn,
session: DbSession,
management_token: ManagementToken,
) -> ShareUpdatedOut:
share = share_service.update_share(
session,
share_token,
management_token=management_token,
title=payload.title,
document=payload.map,
)
return ShareUpdatedOut(
share_token=share.share_token,
share_path=_share_path(share.share_token),
title=share.title,
updated_at=share.updated_at,
)
@router.delete(
"/{share_token}",
status_code=status.HTTP_204_NO_CONTENT,
response_class=Response,
summary="Supprimer le partage (necessite le jeton de gestion)",
)
def delete_share(
share_token: str,
session: DbSession,
management_token: ManagementToken,
) -> Response:
share_service.delete_share(session, share_token, management_token=management_token)
return Response(status_code=status.HTTP_204_NO_CONTENT)

52
backend/app/config.py Normal file
View File

@@ -0,0 +1,52 @@
"""Configuration applicative, entierement pilotee par variables d'environnement."""
from __future__ import annotations
from functools import lru_cache
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""Parametres du backend.
Chaque champ est surchargeable via une variable d'environnement de meme nom
(insensible a la casse) : APP_NAME, DATABASE_URL, MAX_NODES, ...
"""
model_config = SettingsConfigDict(env_file=".env", extra="ignore", case_sensitive=False)
app_name: str = "OpenMind"
app_port: int = 8080
app_env: str = "production"
database_url: str = "sqlite:////app/data/openmind.db"
static_dir: str = "/app/static"
max_map_size_mb: float = Field(default=10.0, gt=0)
max_nodes: int = Field(default=5000, gt=0)
max_title_length: int = Field(default=200, gt=0)
# Liste d'origines autorisees, separees par des virgules.
# Vide = aucune origine externe (l'app est servie par ce meme serveur).
cors_origins: str = ""
log_level: str = "INFO"
@property
def max_map_size_bytes(self) -> int:
return int(self.max_map_size_mb * 1024 * 1024)
@property
def cors_origin_list(self) -> list[str]:
return [origin.strip() for origin in self.cors_origins.split(",") if origin.strip()]
@property
def is_debug(self) -> bool:
return self.app_env.lower() in {"dev", "development", "debug"}
@lru_cache
def get_settings() -> Settings:
return Settings()

View File

9
backend/app/db/base.py Normal file
View File

@@ -0,0 +1,9 @@
"""Base declarative SQLAlchemy commune a tous les modeles."""
from __future__ import annotations
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
"""Classe de base des modeles ORM."""

26
backend/app/db/init_db.py Normal file
View File

@@ -0,0 +1,26 @@
"""Initialisation du schema SQLite.
La V1 n'utilise pas Alembic : le schema tient dans une seule table et
`create_all` suffit. Toute evolution future du schema devra soit passer par
Alembic, soit par une migration explicite ajoutee ici (voir README).
"""
from __future__ import annotations
import logging
from sqlalchemy import Engine
from app.db.base import Base
from app.db.session import engine as default_engine
from app.models import share as _share_model # noqa: F401 (enregistre la table)
logger = logging.getLogger(__name__)
SCHEMA_VERSION = 1
def init_db(engine: Engine | None = None) -> None:
target = engine or default_engine
Base.metadata.create_all(bind=target)
logger.info("Schema SQLite initialise (version %s)", SCHEMA_VERSION)

62
backend/app/db/session.py Normal file
View File

@@ -0,0 +1,62 @@
"""Creation du moteur SQLAlchemy et de la session de requete."""
from __future__ import annotations
import logging
from collections.abc import Generator
from pathlib import Path
from sqlalchemy import Engine, create_engine, event
from sqlalchemy.orm import Session, sessionmaker
from app.config import get_settings
logger = logging.getLogger(__name__)
_SQLITE_FILE_PREFIX = "sqlite:///"
def _ensure_sqlite_directory(database_url: str) -> None:
"""Cree le dossier parent du fichier SQLite s'il n'existe pas encore."""
if not database_url.startswith(_SQLITE_FILE_PREFIX):
return
raw_path = database_url[len(_SQLITE_FILE_PREFIX) :]
if not raw_path or raw_path == ":memory:":
return
path = Path(raw_path)
if path.parent and str(path.parent) not in ("", "."):
path.parent.mkdir(parents=True, exist_ok=True)
def create_db_engine(database_url: str | None = None) -> Engine:
settings = get_settings()
url = database_url or settings.database_url
_ensure_sqlite_directory(url)
connect_args = {"check_same_thread": False} if url.startswith("sqlite") else {}
engine = create_engine(url, connect_args=connect_args, future=True, echo=False)
if url.startswith("sqlite"):
@event.listens_for(engine, "connect")
def _set_sqlite_pragma(dbapi_connection, _connection_record): # type: ignore[no-untyped-def]
cursor = dbapi_connection.cursor()
# WAL : meilleure resistance aux arrets brutaux du conteneur.
cursor.execute("PRAGMA journal_mode=WAL")
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
return engine
engine: Engine = create_db_engine()
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
def get_db() -> Generator[Session, None, None]:
"""Dependance FastAPI : ouvre une session et la referme systematiquement."""
session = SessionLocal()
try:
yield session
finally:
session.close()

View File

@@ -0,0 +1,33 @@
"""Configuration des logs backend : lisible, structure, sans donnee sensible."""
from __future__ import annotations
import logging
import sys
_LOG_FORMAT = "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"
_DATE_FORMAT = "%Y-%m-%dT%H:%M:%S"
# Champs qui ne doivent jamais apparaitre dans les logs.
REDACTED_KEYS = frozenset({"management_token", "managementToken", "authorization"})
def configure_logging(level: str = "INFO") -> None:
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter(_LOG_FORMAT, datefmt=_DATE_FORMAT))
root = logging.getLogger()
root.handlers.clear()
root.addHandler(handler)
root.setLevel(level.upper())
# Uvicorn possede ses propres handlers : on les aligne sur les notres.
for name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
logger = logging.getLogger(name)
logger.handlers.clear()
logger.propagate = True
def safe_extra(data: dict[str, object]) -> dict[str, object]:
"""Retire les valeurs sensibles avant de les envoyer dans un log."""
return {k: ("***" if k in REDACTED_KEYS else v) for k, v in data.items()}

105
backend/app/main.py Normal file
View File

@@ -0,0 +1,105 @@
"""Point d'entree de l'application FastAPI.
Ce fichier se limite a l'assemblage : configuration, middlewares, gestion des
erreurs, montage des routeurs et du frontend statique.
"""
from __future__ import annotations
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from app.api import api_router
from app.config import get_settings
from app.db.init_db import init_db
from app.logging_config import configure_logging
from app.middleware import MaxBodySizeMiddleware, SecurityHeadersMiddleware
from app.services.errors import ShareError
from app.web import mount_frontend
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI): # type: ignore[no-untyped-def]
settings = get_settings()
configure_logging(settings.log_level)
init_db()
logger.info(
"%s demarre (max_nodes=%s, max_map_size_mb=%s)",
settings.app_name,
settings.max_nodes,
settings.max_map_size_mb,
)
yield
logger.info("Arret de %s", settings.app_name)
def create_app() -> FastAPI:
settings = get_settings()
configure_logging(settings.log_level)
app = FastAPI(
title=f"{settings.app_name} API",
version="1.0.0",
description="API de partage de cartes. Les cartes privees restent dans le navigateur.",
lifespan=lifespan,
docs_url="/api/docs",
openapi_url="/api/openapi.json",
redoc_url=None,
)
app.add_middleware(SecurityHeadersMiddleware)
app.add_middleware(MaxBodySizeMiddleware, max_body_size=settings.max_map_size_bytes)
# CORS restrictif : par defaut aucune origine externe, l'app etant servie
# par ce meme serveur (meme origine).
origins = settings.cors_origin_list
if origins:
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=False,
allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
allow_headers=["Content-Type", "X-Management-Token"],
)
register_exception_handlers(app)
app.include_router(api_router)
mount_frontend(app, settings.static_dir)
return app
def register_exception_handlers(app: FastAPI) -> None:
"""Aucune erreur brute ne doit remonter au client."""
@app.exception_handler(ShareError)
async def _share_error(_request: Request, exc: ShareError) -> JSONResponse:
return JSONResponse(status_code=exc.status_code, content={"detail": exc.message})
@app.exception_handler(RequestValidationError)
async def _validation_error(_request: Request, exc: RequestValidationError) -> JSONResponse:
# On expose un message lisible, pas la structure interne de validation.
first = exc.errors()[0] if exc.errors() else None
detail = "Les donnees envoyees ne forment pas une carte valide."
if first and first.get("msg"):
reason = str(first["msg"]).replace("Value error, ", "")
detail = f"{detail} ({reason})"
return JSONResponse(status_code=422, content={"detail": detail})
@app.exception_handler(Exception)
async def _unhandled(request: Request, exc: Exception) -> JSONResponse:
# La traceback complete reste cote serveur.
logger.exception("Erreur non geree sur %s %s", request.method, request.url.path)
return JSONResponse(
status_code=500,
content={"detail": "Une erreur interne est survenue. Reessayez plus tard."},
)
app = create_app()

75
backend/app/middleware.py Normal file
View File

@@ -0,0 +1,75 @@
"""Middlewares HTTP : en-tetes de securite et limite de taille des requetes."""
from __future__ import annotations
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
CONTENT_SECURITY_POLICY = "; ".join(
[
"default-src 'self'",
# Les styles inline sont utilises par React Flow et par l'export image.
"style-src 'self' 'unsafe-inline'",
"script-src 'self'",
"img-src 'self' data: blob:",
"font-src 'self' data:",
"connect-src 'self'",
# Service worker et manifeste de la PWA.
"worker-src 'self'",
"manifest-src 'self'",
"object-src 'none'",
"base-uri 'self'",
"form-action 'self'",
"frame-ancestors 'none'",
]
)
SECURITY_HEADERS = {
"Content-Security-Policy": CONTENT_SECURITY_POLICY,
"X-Content-Type-Options": "nosniff",
"X-Frame-Options": "DENY",
"Referrer-Policy": "no-referrer",
"Permissions-Policy": "geolocation=(), microphone=(), camera=(), interest-cohort=()",
"Cross-Origin-Opener-Policy": "same-origin",
}
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next): # type: ignore[no-untyped-def]
response: Response = await call_next(request)
for header, value in SECURITY_HEADERS.items():
response.headers.setdefault(header, value)
return response
class MaxBodySizeMiddleware(BaseHTTPMiddleware):
"""Refuse proprement les corps de requete trop volumineux.
On s'appuie sur Content-Length : le client est rejete avant toute lecture
complete du corps, et recoit un message clair plutot qu'une erreur brute.
"""
def __init__(self, app, max_body_size: int) -> None: # type: ignore[no-untyped-def]
super().__init__(app)
self.max_body_size = max_body_size
async def dispatch(self, request: Request, call_next): # type: ignore[no-untyped-def]
content_length = request.headers.get("content-length")
if content_length is not None:
try:
size = int(content_length)
except ValueError:
size = 0
if size > self.max_body_size:
limit_mb = self.max_body_size / (1024 * 1024)
return JSONResponse(
status_code=413,
content={
"detail": (
"La carte est trop volumineuse pour etre partagee "
f"(limite : {limit_mb:.0f} Mo)."
)
},
)
return await call_next(request)

View File

@@ -0,0 +1,3 @@
from app.models.share import SharedMap
__all__ = ["SharedMap"]

View File

@@ -0,0 +1,41 @@
"""Modele ORM d'une carte explicitement partagee par son proprietaire."""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import DateTime, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
def utcnow() -> datetime:
return datetime.now(UTC)
class SharedMap(Base):
"""Snapshot serveur d'une carte, cree uniquement sur action de l'utilisateur."""
__tablename__ = "shared_maps"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
# Identifiant public present dans l'URL /s/<share_token>.
share_token: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
# Seul le hash du management token est stocke : le secret n'existe que
# dans l'IndexedDB du proprietaire.
management_token_hash: Mapped[str] = mapped_column(String(64), nullable=False)
title: Mapped[str] = mapped_column(String(255), nullable=False, default="")
map_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
map_data: Mapped[str] = mapped_column(Text, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=utcnow, onupdate=utcnow
)
def __repr__(self) -> str: # pragma: no cover - aide au debug uniquement
return f"<SharedMap id={self.id} share_token={self.share_token!r}>"

View File

@@ -0,0 +1,15 @@
from app.schemas.map import MapDocumentIn
from app.schemas.share import (
ShareCreatedOut,
ShareCreateIn,
SharePublicOut,
ShareUpdatedOut,
)
__all__ = [
"MapDocumentIn",
"ShareCreateIn",
"ShareCreatedOut",
"SharePublicOut",
"ShareUpdatedOut",
]

View File

@@ -0,0 +1,62 @@
"""Validation Pydantic du document carte envoye au serveur.
Le backend ne cherche pas a comprendre finement une carte : il valide sa
structure, ses limites de taille, puis la stocke telle quelle. `extra="allow"`
garantit qu'une version future du format frontend reste acceptee.
"""
from __future__ import annotations
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from app.config import get_settings
MAX_SUPPORTED_MAP_VERSION = 100
class MapNodeIn(BaseModel):
model_config = ConfigDict(extra="allow")
id: str = Field(min_length=1, max_length=128)
text: str = Field(default="", max_length=5000)
class MapEdgeIn(BaseModel):
model_config = ConfigDict(extra="allow")
id: str = Field(min_length=1, max_length=256)
source: str = Field(min_length=1, max_length=128)
target: str = Field(min_length=1, max_length=128)
class MapDocumentIn(BaseModel):
"""Document carte complet, tel que produit par le frontend."""
model_config = ConfigDict(extra="allow")
version: int = Field(ge=1, le=MAX_SUPPORTED_MAP_VERSION)
id: str = Field(min_length=1, max_length=128)
name: str = Field(default="", max_length=255)
nodes: list[MapNodeIn] = Field(default_factory=list)
edges: list[MapEdgeIn] = Field(default_factory=list)
viewport: dict[str, Any] = Field(default_factory=dict)
settings: dict[str, Any] = Field(default_factory=dict)
@field_validator("nodes")
@classmethod
def _check_node_count(cls, value: list[MapNodeIn]) -> list[MapNodeIn]:
max_nodes = get_settings().max_nodes
if len(value) > max_nodes:
raise ValueError(
f"La carte depasse la limite de {max_nodes} noeuds ({len(value)} noeuds fournis)."
)
return value
@model_validator(mode="after")
def _check_edge_count(self) -> MapDocumentIn:
max_edges = get_settings().max_nodes * 4
if len(self.edges) > max_edges:
raise ValueError(f"La carte depasse la limite de {max_edges} liens.")
return self

View File

@@ -0,0 +1,51 @@
"""Schemas d'entree/sortie de l'API de partage."""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
from app.schemas.map import MapDocumentIn
class ShareCreateIn(BaseModel):
"""Corps de POST /api/shares et de PUT /api/shares/{token}."""
model_config = ConfigDict(extra="forbid")
title: str = Field(default="", max_length=255)
map: MapDocumentIn
class ShareCreatedOut(BaseModel):
"""Reponse renvoyee une seule fois, a la creation du partage.
`management_token` n'est jamais renvoye ailleurs et n'apparait jamais
dans une URL publique : il est conserve par le proprietaire.
"""
share_token: str
management_token: str
share_path: str
title: str
created_at: datetime
updated_at: datetime
class ShareUpdatedOut(BaseModel):
share_token: str
share_path: str
title: str
updated_at: datetime
class SharePublicOut(BaseModel):
"""Vue publique : aucune information de gestion."""
share_token: str
title: str
map_version: int
map: dict
created_at: datetime
updated_at: datetime

View File

View File

@@ -0,0 +1,34 @@
"""Erreurs metier du domaine partage.
Elles sont converties en reponses HTTP propres par les handlers de `app.main` :
aucune trace technique n'est exposee au client.
"""
from __future__ import annotations
class ShareError(Exception):
"""Erreur metier generique."""
status_code = 400
message = "Requete invalide."
def __init__(self, message: str | None = None) -> None:
super().__init__(message or self.message)
if message:
self.message = message
class ShareNotFoundError(ShareError):
status_code = 404
message = "Ce partage n'existe pas ou a ete supprime."
class ShareForbiddenError(ShareError):
status_code = 403
message = "Jeton de gestion invalide."
class SharePayloadTooLargeError(ShareError):
status_code = 413
message = "La carte depasse la taille maximale autorisee."

View File

@@ -0,0 +1,131 @@
"""Logique metier du partage de cartes."""
from __future__ import annotations
import json
import logging
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.config import get_settings
from app.models.share import SharedMap, utcnow
from app.schemas.map import MapDocumentIn
from app.services.errors import (
ShareForbiddenError,
ShareNotFoundError,
SharePayloadTooLargeError,
)
from app.services.tokens import (
generate_management_token,
generate_share_token,
hash_token,
verify_token,
)
logger = logging.getLogger(__name__)
def serialize_map(document: MapDocumentIn) -> str:
"""Serialise le document carte en JSON compact et deterministe."""
return json.dumps(
document.model_dump(mode="json"),
ensure_ascii=False,
separators=(",", ":"),
)
def _guard_payload_size(payload: str) -> None:
settings = get_settings()
size = len(payload.encode("utf-8"))
if size > settings.max_map_size_bytes:
raise SharePayloadTooLargeError(
f"La carte fait {size / (1024 * 1024):.1f} Mo, "
f"la limite est de {settings.max_map_size_mb} Mo."
)
def _get_by_token(session: Session, share_token: str) -> SharedMap:
stmt = select(SharedMap).where(SharedMap.share_token == share_token)
share = session.execute(stmt).scalar_one_or_none()
if share is None:
raise ShareNotFoundError()
return share
def create_share(session: Session, *, title: str, document: MapDocumentIn) -> tuple[SharedMap, str]:
"""Cree un partage et retourne (entite, management_token en clair).
Le management token en clair n'est renvoye qu'ici, une seule fois.
"""
payload = serialize_map(document)
_guard_payload_size(payload)
management_token = generate_management_token()
share = SharedMap(
share_token=generate_share_token(),
management_token_hash=hash_token(management_token),
title=(title or document.name or "Carte sans titre")[: get_settings().max_title_length],
map_version=document.version,
map_data=payload,
)
session.add(share)
session.commit()
session.refresh(share)
logger.info("Partage cree share_token=%s nodes=%s", share.share_token, len(document.nodes))
return share, management_token
def get_share(session: Session, share_token: str) -> SharedMap:
return _get_by_token(session, share_token)
def load_map_data(share: SharedMap) -> dict:
"""Deserialise le snapshot stocke. Ne fait qu'un json.loads : aucun code execute."""
try:
data = json.loads(share.map_data)
except json.JSONDecodeError: # pragma: no cover - donnee corrompue
logger.exception("Snapshot illisible pour share_token=%s", share.share_token)
raise ShareNotFoundError("Le contenu de ce partage est illisible.") from None
if not isinstance(data, dict): # pragma: no cover - donnee corrompue
raise ShareNotFoundError("Le contenu de ce partage est illisible.")
return data
def update_share(
session: Session,
share_token: str,
*,
management_token: str,
title: str,
document: MapDocumentIn,
) -> SharedMap:
share = _get_by_token(session, share_token)
if not verify_token(management_token, share.management_token_hash):
logger.warning("Tentative de mise a jour refusee share_token=%s", share_token)
raise ShareForbiddenError()
payload = serialize_map(document)
_guard_payload_size(payload)
share.title = (title or document.name or share.title)[: get_settings().max_title_length]
share.map_version = document.version
share.map_data = payload
share.updated_at = utcnow()
session.commit()
session.refresh(share)
logger.info("Partage mis a jour share_token=%s", share_token)
return share
def delete_share(session: Session, share_token: str, *, management_token: str) -> None:
share = _get_by_token(session, share_token)
if not verify_token(management_token, share.management_token_hash):
logger.warning("Tentative de suppression refusee share_token=%s", share_token)
raise ShareForbiddenError()
session.delete(share)
session.commit()
logger.info("Partage supprime share_token=%s", share_token)

View File

@@ -0,0 +1,45 @@
"""Generation et verification des jetons de partage.
Deux secrets distincts sont produits a la creation d'un partage :
* ``share_token`` : identifiant public, present dans /s/<token>, lecture seule.
* ``management_token`` : secret de gestion (PUT/DELETE), jamais dans une URL,
stocke uniquement dans l'IndexedDB du proprietaire.
Seul le hash SHA-256 du management token est conserve en base.
"""
from __future__ import annotations
import hashlib
import secrets
# 12 octets -> 16 caracteres url-safe, ~96 bits d'entropie : non enumerable.
SHARE_TOKEN_BYTES = 12
# 32 octets -> ~256 bits d'entropie pour le secret de gestion.
MANAGEMENT_TOKEN_BYTES = 32
def generate_share_token() -> str:
return secrets.token_urlsafe(SHARE_TOKEN_BYTES)
def generate_management_token() -> str:
return secrets.token_urlsafe(MANAGEMENT_TOKEN_BYTES)
def hash_token(token: str) -> str:
"""Hash du secret de gestion.
Le token est deja un secret aleatoire de haute entropie : un SHA-256 simple
suffit (pas de risque de force brute par dictionnaire, contrairement a un
mot de passe choisi par un humain).
"""
return hashlib.sha256(token.encode("utf-8")).hexdigest()
def verify_token(token: str, expected_hash: str) -> bool:
"""Comparaison a temps constant."""
if not token or not expected_hash:
return False
return secrets.compare_digest(hash_token(token), expected_hash)

76
backend/app/web.py Normal file
View File

@@ -0,0 +1,76 @@
"""Service des fichiers statiques du frontend (SPA) par FastAPI.
Le build Vite est copie dans `settings.static_dir`. Toute route qui n'est ni
/api/... ni un fichier existant retourne index.html, afin que le routeur
cote client puisse gerer /map/:id et /s/:token.
"""
from __future__ import annotations
import logging
from pathlib import Path
from fastapi import APIRouter, FastAPI, Request
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
logger = logging.getLogger(__name__)
_PLACEHOLDER = """<!doctype html>
<html lang="fr"><head><meta charset="utf-8"><title>Frontend non compile</title></head>
<body style="font-family:system-ui;padding:2rem;max-width:40rem;margin:auto">
<h1>Frontend non compile</h1>
<p>L'API est disponible sur <code>/api/health</code>.</p>
<p>En developpement, lancez <code>npm run dev</code> dans <code>frontend/</code>.</p>
</body></html>"""
# Fichiers servis a la racine et qui ne doivent pas etre mis en cache longtemps.
_NO_CACHE_FILES = {"index.html", "sw.js", "manifest.webmanifest"}
def _safe_resolve(static_dir: Path, relative_path: str) -> Path | None:
"""Empeche toute sortie du dossier statique (path traversal)."""
candidate = (static_dir / relative_path.lstrip("/")).resolve()
try:
candidate.relative_to(static_dir.resolve())
except ValueError:
return None
return candidate
def mount_frontend(app: FastAPI, static_dir_path: str) -> None:
static_dir = Path(static_dir_path)
router = APIRouter(include_in_schema=False)
if (static_dir / "assets").is_dir():
# Les assets Vite sont hashes : cache long sans risque.
app.mount(
"/assets",
StaticFiles(directory=static_dir / "assets"),
name="assets",
)
@router.api_route("/{full_path:path}", methods=["GET", "HEAD"])
def serve_spa(full_path: str, request: Request): # type: ignore[no-untyped-def]
if full_path.startswith("api/"):
return JSONResponse(status_code=404, content={"detail": "Ressource introuvable."})
if not static_dir.is_dir():
return HTMLResponse(_PLACEHOLDER, status_code=200)
if full_path:
candidate = _safe_resolve(static_dir, full_path)
if candidate is not None and candidate.is_file():
headers = (
{"Cache-Control": "no-cache"}
if candidate.name in _NO_CACHE_FILES
else {"Cache-Control": "public, max-age=3600"}
)
return FileResponse(candidate, headers=headers)
index = static_dir / "index.html"
if index.is_file():
return FileResponse(index, headers={"Cache-Control": "no-cache"})
return HTMLResponse(_PLACEHOLDER, status_code=200)
app.include_router(router)