Separer le sysadmin instance des groupes

This commit is contained in:
Daniel Allaire 2026-06-01 07:22:01 -04:00
parent 7ed9dd9167
commit bcc2552c44
8 changed files with 102 additions and 28 deletions

View file

@ -21,10 +21,18 @@ def verify_pin(pin: str, hashed: str) -> bool:
return bcrypt.checkpw(pin.encode("utf-8"), hashed.encode("utf-8"))
def create_token(membre_id: str, groupe_id: str | None, expire_jours: int = None, admin_id: str | None = None) -> str:
def create_token(
membre_id: str,
groupe_id: str | None,
expire_jours: int = None,
admin_id: str | None = None,
subject_type: str = "membre",
) -> str:
jours = expire_jours or settings.JWT_EXPIRE_DAYS
expire = datetime.now(timezone.utc) + timedelta(days=jours)
payload = {"sub": str(membre_id), "exp": expire}
if subject_type != "membre":
payload["typ"] = subject_type
if groupe_id:
payload["grp"] = str(groupe_id)
if admin_id:
@ -48,11 +56,18 @@ async def get_current_membre(
db: AsyncSession = Depends(get_db),
):
from app.models.membre import Membre
from app.models.instance_admin import InstanceAdmin
payload = decode_token(credentials.credentials)
membre_id = payload.get("sub")
groupe_id = payload.get("grp")
if not membre_id:
raise HTTPException(status_code=401, detail="Token invalide")
if payload.get("typ") == "instance_admin":
result = await db.execute(select(InstanceAdmin).where(InstanceAdmin.id == membre_id, InstanceAdmin.actif == True))
admin = result.scalar_one_or_none()
if not admin:
raise HTTPException(status_code=401, detail="Sysadmin d'instance introuvable ou inactif")
return admin
result = await db.execute(select(Membre).where(Membre.id == membre_id, Membre.actif == True))
membre = result.scalar_one_or_none()
if not membre:

View file

@ -29,6 +29,7 @@ from app.models.destinataire_contribution import DestinataireContribution # noq
from app.models.rapport_adopte import RapportAdopte # noqa: F401
from app.models.vente_jeton import VenteJeton # noqa: F401
from app.models.historique_modification import HistoriqueModification # noqa: F401
from app.models.instance_admin import InstanceAdmin # noqa: F401
@asynccontextmanager

View file

@ -0,0 +1,30 @@
"""Administrateurs globaux de l'instance."""
import uuid
from datetime import datetime, timezone
from sqlalchemy import String, Boolean, DateTime
from sqlalchemy.orm import Mapped, mapped_column
from app.core.database import Base
class InstanceAdmin(Base):
__tablename__ = "instance_admins"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
identifiant: Mapped[str] = mapped_column(String(80), unique=True, nullable=False)
prenom: Mapped[str] = mapped_column(String(50), nullable=False, default="Sysadmin")
pin_hash: Mapped[str] = mapped_column(String(255), nullable=False)
actif: Mapped[bool] = mapped_column(Boolean, default=True)
cree_le: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
modifie_le: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
@property
def courriel(self) -> str:
return self.identifiant
@property
def groupe_id(self):
return None
@property
def is_sysadmin(self) -> bool:
return True

View file

@ -7,6 +7,7 @@ from pydantic import BaseModel
from app.core.database import get_db
from app.core.security import create_token, get_current_membre, is_sysadmin, verify_pin
from app.services.admin_database import (
assurer_sysadmin_instance,
creer_groupe,
initialiser_demo,
modifier_groupe,
@ -45,23 +46,22 @@ async def require_sysadmin(membre=Depends(get_current_membre)):
@router.post("/sysadmin-login")
async def sysadmin_login(req: SysadminLoginRequest, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(Membre).where(Membre.courriel == "sysadmin", Membre.actif == True))
for membre in result.scalars().all():
if verify_pin(req.pin, membre.pin_hash):
token = create_token(str(membre.id), None, expire_jours=1)
return {
"access_token": token,
"token_type": "bearer",
"sysadmin_instance": True,
"membre": {
"id": str(membre.id),
"prenom": "Sysadmin",
"courriel": "sysadmin",
"groupe_id": None,
"date_abstinence": None,
"is_sysadmin": True,
},
}
admin = await assurer_sysadmin_instance(db)
if admin.actif and verify_pin(req.pin, admin.pin_hash):
token = create_token(str(admin.id), None, expire_jours=1, subject_type="instance_admin")
return {
"access_token": token,
"token_type": "bearer",
"sysadmin_instance": True,
"membre": {
"id": str(admin.id),
"prenom": admin.prenom,
"courriel": admin.identifiant,
"groupe_id": None,
"date_abstinence": None,
"is_sysadmin": True,
},
}
raise HTTPException(status_code=401, detail="PIN Sysadmin invalide")

View file

@ -20,6 +20,7 @@ from app.models.envoi_contribution import EnvoiContribution
from app.models.evenement import Evenement
from app.models.groupe import Groupe
from app.models.historique_modification import HistoriqueModification
from app.models.instance_admin import InstanceAdmin
from app.models.jeton import Jeton
from app.models.journal_action import JournalAction
from app.models.litterature import Litterature
@ -42,6 +43,7 @@ from app.models.vente_jeton import VenteJeton
DEMO_GROUP_NAME = "Groupe Démonstration"
OLD_DEMO_GROUP_NAMES = ("Groupe Résilience",)
INSTANCE_SYSADMIN_IDENTIFIANT = "sysadmin"
POSTES = [
@ -73,6 +75,24 @@ MEMBRES_DEMO = [
]
async def assurer_sysadmin_instance(db: AsyncSession) -> InstanceAdmin:
result = await db.execute(
select(InstanceAdmin).where(InstanceAdmin.identifiant == INSTANCE_SYSADMIN_IDENTIFIANT)
)
admin = result.scalar_one_or_none()
if admin:
return admin
admin = InstanceAdmin(
identifiant=INSTANCE_SYSADMIN_IDENTIFIANT,
prenom="Sysadmin",
pin_hash=hash_pin("0000"),
actif=True,
)
db.add(admin)
await db.flush()
return admin
def _catalogue_litterature() -> list[dict]:
chemin = Path(__file__).resolve().parents[3] / "data" / "import" / "Liste-de-prix-rev.-Fevrier-2026.csv"
if chemin.exists():

View file

@ -16,6 +16,7 @@ from app.models.collecte import Collecte
from app.models.depense import Depense
from app.models.evenement import Evenement
from app.models.jeton import Jeton
from app.models.instance_admin import InstanceAdmin
from app.models.litterature import Litterature
from app.models.presence import Presence
from app.models.proposition import Proposition
@ -63,9 +64,10 @@ class AuditTests(unittest.TestCase):
class SecurityTokenTests(unittest.TestCase):
def test_sysadmin_instance_token_has_no_group_context(self):
token = create_token("member-id", None, expire_jours=1)
token = create_token("admin-id", None, expire_jours=1, subject_type="instance_admin")
payload = decode_token(token)
self.assertEqual(payload["sub"], "member-id")
self.assertEqual(payload["sub"], "admin-id")
self.assertEqual(payload["typ"], "instance_admin")
self.assertNotIn("grp", payload)
@ -74,6 +76,10 @@ class MultiTenantTests(unittest.TestCase):
self.assertIn("groupe_id", Poste.__table__.columns)
self.assertFalse(Poste.__table__.columns["groupe_id"].nullable)
def test_instance_admin_is_not_group_scoped(self):
self.assertNotIn("groupe_id", InstanceAdmin.__table__.columns)
self.assertTrue(InstanceAdmin.__table__.columns["identifiant"].unique)
def test_core_business_tables_are_group_scoped(self):
for model in [
Affectation,

View file

@ -11,7 +11,7 @@ import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))
from app.core.database import engine, Base, async_session
from app.services.admin_database import initialiser_demo
from app.services.admin_database import assurer_sysadmin_instance, initialiser_demo
# Importer TOUS les modèles pour que create_all crée toutes les tables
from app.models.groupe import Groupe
from app.models.membre import Membre
@ -40,6 +40,7 @@ from app.models.reserve import Reserve
from app.models.transaction_comptable import TransactionComptable
from app.models.mouvement_reserve import MouvementReserve
from app.models.destinataire_contribution import DestinataireContribution
from app.models.instance_admin import InstanceAdmin
async def seed():
@ -47,6 +48,7 @@ async def seed():
await conn.run_sync(Base.metadata.create_all)
async with async_session() as db:
await assurer_sysadmin_instance(db)
demo = await initialiser_demo(db)
await db.commit()

View file

@ -84,13 +84,6 @@ export default function Login() {
<p className="text-blue-200 text-sm mt-1">District 87-16</p>
</div>
<a
href="/presentation.html"
className="mb-4 block w-full rounded-xl bg-white/95 p-4 text-center text-sm font-bold text-aa-blue shadow-lg active:scale-[0.98]"
>
Découvrir l'application
</a>
<div className="flex-1 space-y-3 pb-6">
{groupes.length === 0 ? (
<div className="rounded-2xl bg-white/10 p-5 text-center text-blue-100">
@ -187,6 +180,13 @@ export default function Login() {
})}
</div>
<a
href="/presentation.html"
className="mb-4 block w-full rounded-xl bg-white/95 p-4 text-center text-sm font-bold text-aa-blue shadow-lg active:scale-[0.98]"
>
Découvrir l'application
</a>
<button
type="button"
onClick={ouvrirSysadmin}