Phase 3 : hardening qualite de code - typage strict, securite, dead code, a11y
Config strictement stricte partout (ruff, mypy --strict, bandit, vulture, import-linter, eslint, stylelint), aucune regle desactivee "pour ne pas casser le build" - l'existant a ete corrige pour la satisfaire plutot que l'inverse. Hooks pre-commit locaux (language: system) bloquants. - Typage mypy --strict propage a tout le moteur (db, screens, auth, core, ai, routes, puis publish/scripts/tests/app.py/build_css.py). - Securite : fuite de handle fichier Windows corrigee dans l'export SCORM (routes/publish/export_scorm.py), CSRF/RNG non-crypto/xAPI documentes (# nosec, # NOSONAR justifies), nouveau db.json_for_script() (echappe "</script>" dans le JSON embarque en <script>, 25 sites). - Architecture : imports circulaires/F811 nettoyes, contrats import-linter respectes, code mort retire (vulture). - Accessibilite : 69 champs de formulaire sans label correctement associe corriges (for/id ou aria-label) sur 11 templates. - ESLint/Stylelint : lot mecanique JS/CSS, regles ajustees puis appliquees (aucune desactivee sans verification individuelle). - Tests : isolation du compte admin partage (nettoyage ponctuel + fixture de teardown automatique en filet de securite), suite complete verte (591 tests Python, 241 tests JS). - SonarQube Community Build self-heberge (Docker + PostgreSQL) : rapport complet analyse point par point, faux positifs documentes. - .gitattributes ajoute (LF force) : core.autocrlf=true sur cette machine faisait echouer ESLint (linebreak-style) via un bug connu de git (checkout "en place" qui ignore l'eol force sur un fichier deja present sur disque - contourne en supprimant puis recreant chaque fichier suivi). djLint (H021, styles inline) volontairement saute pour ce commit - backlog assume, deja documente, traite dans un lot separe. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 5
parent
7db4803b93
commit
c57420c8c9
+82
-51
@@ -4,63 +4,94 @@ create_user.py pour le détail des règles. Base SQLite entièrement séparée
|
||||
de db/ (une base par JEU) : ces comptes n'appartiennent à aucun jeu, ils
|
||||
en POSSÈDENT un (project_slug)."""
|
||||
|
||||
from .confirm_totp import confirm_totp
|
||||
from .connection import users_db_path
|
||||
from .count_admins import count_admins
|
||||
from .create_user import UserCreationError, create_user
|
||||
from .create_user_asset import create_user_asset
|
||||
from .delete_user import delete_user
|
||||
from .delete_user_asset import delete_user_asset
|
||||
from .ensure_schema import ensure_users_schema
|
||||
from .is_first_user import is_first_user
|
||||
from .create_user import create_user, UserCreationError
|
||||
from .ensure_user_assets_schema import ensure_user_assets_schema
|
||||
from .get_user_asset import get_user_asset
|
||||
from .get_user_by_email import get_user_by_email
|
||||
from .get_user_by_id import get_user_by_id
|
||||
from .verify_password import verify_password
|
||||
from .confirm_totp import confirm_totp
|
||||
from .verify_totp import verify_totp
|
||||
from .set_project_slug import set_project_slug
|
||||
from .password_strength import password_strength, MIN_SCORE_REQUIRED
|
||||
from .totp_qrcode_svg import totp_provisioning_uri, totp_qrcode_svg
|
||||
from .rate_limit import lockout_minutes_for, seconds_locked_remaining, lockout_message
|
||||
from .record_failed_attempt import record_failed_attempt
|
||||
from .reset_failed_attempts import reset_failed_attempts
|
||||
from .recovery_codes import generate_recovery_codes, verify_recovery_code
|
||||
from .set_password import set_password
|
||||
from .password_reset import (
|
||||
create_password_reset_token, get_user_id_for_valid_token, consume_password_reset_token,
|
||||
TOKEN_TTL_MINUTES,
|
||||
)
|
||||
from .send_email import send_password_reset_email, EmailNotConfiguredError
|
||||
from .update_profile import update_profile
|
||||
from .count_admins import count_admins
|
||||
from .delete_user import delete_user
|
||||
from .update_email import update_email, EmailUpdateError
|
||||
from .ensure_user_assets_schema import ensure_user_assets_schema
|
||||
from .user_assets_dir import user_assets_dir
|
||||
from .create_user_asset import create_user_asset
|
||||
from .list_user_assets import list_user_assets
|
||||
from .get_user_asset import get_user_asset
|
||||
from .delete_user_asset import delete_user_asset
|
||||
from .save_user_asset import save_user_asset
|
||||
from .user_asset_kind import user_asset_kind
|
||||
from .validate_audio_duration import validate_audio_duration, MAX_AUDIO_SECONDS
|
||||
from .validate_video_duration import validate_video_duration, MAX_VIDEO_SECONDS
|
||||
from .image_dimensions import image_dimensions
|
||||
from .is_first_user import is_first_user
|
||||
from .list_user_assets import list_user_assets
|
||||
from .password_reset import (
|
||||
TOKEN_TTL_MINUTES,
|
||||
consume_password_reset_token,
|
||||
create_password_reset_token,
|
||||
get_user_id_for_valid_token,
|
||||
)
|
||||
from .password_strength import MIN_SCORE_REQUIRED, password_strength
|
||||
from .rate_limit import lockout_message, lockout_minutes_for, seconds_locked_remaining
|
||||
from .record_failed_attempt import record_failed_attempt
|
||||
from .recovery_codes import generate_recovery_codes, verify_recovery_code
|
||||
from .reset_failed_attempts import reset_failed_attempts
|
||||
from .save_user_asset import save_user_asset
|
||||
from .send_email import EmailNotConfiguredError, send_password_reset_email
|
||||
from .set_password import set_password
|
||||
from .set_project_slug import set_project_slug
|
||||
from .totp_qrcode_svg import totp_provisioning_uri, totp_qrcode_svg
|
||||
from .update_email import EmailUpdateError, update_email
|
||||
from .update_profile import update_profile
|
||||
from .update_user_asset_scene_kind import update_user_asset_scene_kind
|
||||
from .user_asset_kind import user_asset_kind
|
||||
from .user_assets_dir import user_assets_dir
|
||||
from .validate_audio_duration import MAX_AUDIO_SECONDS, validate_audio_duration
|
||||
from .validate_video_duration import MAX_VIDEO_SECONDS, validate_video_duration
|
||||
from .verify_password import verify_password
|
||||
from .verify_totp import verify_totp
|
||||
|
||||
__all__ = [
|
||||
"users_db_path", "ensure_users_schema", "is_first_user",
|
||||
"create_user", "UserCreationError",
|
||||
"get_user_by_email", "get_user_by_id",
|
||||
"verify_password", "confirm_totp", "verify_totp", "set_project_slug",
|
||||
"password_strength", "MIN_SCORE_REQUIRED",
|
||||
"totp_provisioning_uri", "totp_qrcode_svg",
|
||||
"lockout_minutes_for", "seconds_locked_remaining", "lockout_message",
|
||||
"record_failed_attempt", "reset_failed_attempts",
|
||||
"generate_recovery_codes", "verify_recovery_code",
|
||||
"set_password", "create_password_reset_token", "get_user_id_for_valid_token",
|
||||
"consume_password_reset_token", "TOKEN_TTL_MINUTES",
|
||||
"send_password_reset_email", "EmailNotConfiguredError",
|
||||
"update_profile", "count_admins", "delete_user",
|
||||
"update_email", "EmailUpdateError",
|
||||
"ensure_user_assets_schema", "user_assets_dir", "create_user_asset",
|
||||
"list_user_assets", "get_user_asset", "delete_user_asset", "save_user_asset", "user_asset_kind",
|
||||
"validate_audio_duration", "MAX_AUDIO_SECONDS",
|
||||
"validate_video_duration", "MAX_VIDEO_SECONDS",
|
||||
"image_dimensions", "update_user_asset_scene_kind",
|
||||
"users_db_path",
|
||||
"ensure_users_schema",
|
||||
"is_first_user",
|
||||
"create_user",
|
||||
"UserCreationError",
|
||||
"get_user_by_email",
|
||||
"get_user_by_id",
|
||||
"verify_password",
|
||||
"confirm_totp",
|
||||
"verify_totp",
|
||||
"set_project_slug",
|
||||
"password_strength",
|
||||
"MIN_SCORE_REQUIRED",
|
||||
"totp_provisioning_uri",
|
||||
"totp_qrcode_svg",
|
||||
"lockout_minutes_for",
|
||||
"seconds_locked_remaining",
|
||||
"lockout_message",
|
||||
"record_failed_attempt",
|
||||
"reset_failed_attempts",
|
||||
"generate_recovery_codes",
|
||||
"verify_recovery_code",
|
||||
"set_password",
|
||||
"create_password_reset_token",
|
||||
"get_user_id_for_valid_token",
|
||||
"consume_password_reset_token",
|
||||
"TOKEN_TTL_MINUTES",
|
||||
"send_password_reset_email",
|
||||
"EmailNotConfiguredError",
|
||||
"update_profile",
|
||||
"count_admins",
|
||||
"delete_user",
|
||||
"update_email",
|
||||
"EmailUpdateError",
|
||||
"ensure_user_assets_schema",
|
||||
"user_assets_dir",
|
||||
"create_user_asset",
|
||||
"list_user_assets",
|
||||
"get_user_asset",
|
||||
"delete_user_asset",
|
||||
"save_user_asset",
|
||||
"user_asset_kind",
|
||||
"validate_audio_duration",
|
||||
"MAX_AUDIO_SECONDS",
|
||||
"validate_video_duration",
|
||||
"MAX_VIDEO_SECONDS",
|
||||
"image_dimensions",
|
||||
"update_user_asset_scene_kind",
|
||||
]
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from .connection import connect
|
||||
|
||||
|
||||
def confirm_totp(user_id):
|
||||
def confirm_totp(user_id: int) -> None:
|
||||
conn = connect()
|
||||
conn.execute("UPDATE _users SET totp_confirmed = 1 WHERE id = ?", (user_id,))
|
||||
conn.commit()
|
||||
|
||||
+3
-2
@@ -8,6 +8,7 @@ tests/conftest.py positionne FORGE_USERS_DB_PATH vers un fichier temporaire
|
||||
avant de faire tourner la suite, pour ne jamais toucher à la vraie base de
|
||||
comptes (data/users.db) ni dépendre d'un état "premier compte = admin" déjà
|
||||
consommé par un run précédent."""
|
||||
|
||||
import os
|
||||
import sqlite3
|
||||
|
||||
@@ -15,11 +16,11 @@ _BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
DEFAULT_USERS_DB_PATH = os.path.join(_BASE_DIR, "data", "users.db")
|
||||
|
||||
|
||||
def users_db_path():
|
||||
def users_db_path() -> str:
|
||||
return os.environ.get("FORGE_USERS_DB_PATH") or DEFAULT_USERS_DB_PATH
|
||||
|
||||
|
||||
def connect():
|
||||
def connect() -> sqlite3.Connection:
|
||||
path = users_db_path()
|
||||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||
conn = sqlite3.connect(path, timeout=10)
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
from .connection import connect
|
||||
|
||||
|
||||
def count_admins():
|
||||
def count_admins() -> int:
|
||||
conn = connect()
|
||||
n = conn.execute("SELECT COUNT(*) AS n FROM _users WHERE role = 'admin'").fetchone()["n"]
|
||||
n = int(conn.execute("SELECT COUNT(*) AS n FROM _users WHERE role = 'admin'").fetchone()["n"])
|
||||
conn.close()
|
||||
return n
|
||||
|
||||
+4
-3
@@ -1,9 +1,9 @@
|
||||
import pyotp
|
||||
from werkzeug.security import generate_password_hash
|
||||
|
||||
from .ensure_schema import ensure_users_schema
|
||||
from .connection import connect
|
||||
from .email_validation import is_valid_email
|
||||
from .ensure_schema import ensure_users_schema
|
||||
from .get_user_by_email import get_user_by_email
|
||||
from .is_first_user import is_first_user
|
||||
from .password_strength import password_strength
|
||||
@@ -14,7 +14,7 @@ class UserCreationError(Exception):
|
||||
dans le formulaire d'inscription) — jamais un détail SQL/technique."""
|
||||
|
||||
|
||||
def create_user(email, password, nom, prenom):
|
||||
def create_user(email: str | None, password: str | None, nom: str | None, prenom: str | None) -> int:
|
||||
"""Crée un compte : mot de passe fort (auth/password_strength.py) et
|
||||
2FA (TOTP) rendus obligatoires — le secret est généré ici mais
|
||||
totp_confirmed reste à 0 tant que confirm_totp() n'a pas vérifié un
|
||||
@@ -26,6 +26,7 @@ def create_user(email, password, nom, prenom):
|
||||
email = (email or "").strip().lower()
|
||||
nom = (nom or "").strip()
|
||||
prenom = (prenom or "").strip()
|
||||
password = password or ""
|
||||
if not is_valid_email(email):
|
||||
raise UserCreationError("Adresse email invalide.")
|
||||
if not nom or not prenom:
|
||||
@@ -44,7 +45,7 @@ def create_user(email, password, nom, prenom):
|
||||
VALUES (?, ?, ?, ?, ?, ?, 0)""",
|
||||
(email, generate_password_hash(password), nom, prenom, role, totp_secret),
|
||||
)
|
||||
user_id = conn.execute("SELECT last_insert_rowid() AS id").fetchone()["id"]
|
||||
user_id = int(conn.execute("SELECT last_insert_rowid() AS id").fetchone()["id"])
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return user_id
|
||||
|
||||
@@ -2,7 +2,13 @@ from .connection import connect
|
||||
from .ensure_user_assets_schema import ensure_user_assets_schema
|
||||
|
||||
|
||||
def create_user_asset(user_id, filename, original_name=None, source="upload", scene_kind=None):
|
||||
def create_user_asset(
|
||||
user_id: int,
|
||||
filename: str,
|
||||
original_name: str | None = None,
|
||||
source: str = "upload",
|
||||
scene_kind: str | None = None,
|
||||
) -> int:
|
||||
"""Enregistre une ligne "Mes assets" pour un fichier déjà écrit sur
|
||||
disque (voir user_assets_dir.py) — le fichier lui-même est écrit par
|
||||
l'appelant (route d'upload, ou plus tard la génération IA), cette
|
||||
@@ -21,7 +27,7 @@ def create_user_asset(user_id, filename, original_name=None, source="upload", sc
|
||||
"INSERT INTO _user_assets (user_id, filename, original_name, source, scene_kind) VALUES (?, ?, ?, ?, ?)",
|
||||
(user_id, filename, original_name, source, scene_kind),
|
||||
)
|
||||
asset_id = conn.execute("SELECT last_insert_rowid() AS id").fetchone()["id"]
|
||||
asset_id = int(conn.execute("SELECT last_insert_rowid() AS id").fetchone()["id"])
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return asset_id
|
||||
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
from .connection import connect
|
||||
|
||||
|
||||
def delete_user(user_id):
|
||||
def delete_user(user_id: int) -> None:
|
||||
"""Supprime le compte et tout ce qui lui est directement rattaché
|
||||
(codes de récupération, jetons de réinitialisation encore en cours) —
|
||||
le dossier de projet, lui, est géré par l'appelant (voir
|
||||
|
||||
@@ -5,16 +5,14 @@ from .ensure_user_assets_schema import ensure_user_assets_schema
|
||||
from .user_assets_dir import user_assets_dir
|
||||
|
||||
|
||||
def delete_user_asset(asset_id, user_id):
|
||||
def delete_user_asset(asset_id: int, user_id: int) -> bool:
|
||||
"""N'efface que si asset_id APPARTIENT à user_id (jamais fournie par
|
||||
l'appelant sans vérification — voir get_user_asset.py) : renvoie False
|
||||
sans rien faire sinon, plutôt que de lever une erreur, même patron que
|
||||
screens.delete_scene_object pour un id introuvable."""
|
||||
ensure_user_assets_schema()
|
||||
conn = connect()
|
||||
row = conn.execute(
|
||||
"SELECT filename FROM _user_assets WHERE id = ? AND user_id = ?", (asset_id, user_id)
|
||||
).fetchone()
|
||||
row = conn.execute("SELECT filename FROM _user_assets WHERE id = ? AND user_id = ?", (asset_id, user_id)).fetchone()
|
||||
if not row:
|
||||
conn.close()
|
||||
return False
|
||||
|
||||
@@ -3,5 +3,5 @@ import re
|
||||
EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
|
||||
|
||||
|
||||
def is_valid_email(email):
|
||||
def is_valid_email(email: str | None) -> bool:
|
||||
return bool(EMAIL_RE.match((email or "").strip()))
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from .connection import connect
|
||||
|
||||
|
||||
def ensure_users_schema():
|
||||
def ensure_users_schema() -> None:
|
||||
conn = connect()
|
||||
conn.executescript(
|
||||
"""
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
from .connection import connect
|
||||
|
||||
|
||||
def ensure_user_assets_schema():
|
||||
""""Mes assets" (voir user_assets_dir.py) : une image appartient à un
|
||||
def ensure_user_assets_schema() -> None:
|
||||
""" "Mes assets" (voir user_assets_dir.py) : une image appartient à un
|
||||
COMPTE, jamais à un projet — table dans la base de comptes partagée
|
||||
(auth/connection.py), pas dans le game.db d'un jeu (db/connection.py),
|
||||
pour rester utilisable d'un projet à l'autre du même compte."""
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
from typing import Any
|
||||
|
||||
from .connection import connect
|
||||
from .ensure_user_assets_schema import ensure_user_assets_schema
|
||||
|
||||
|
||||
def get_user_asset(asset_id):
|
||||
def get_user_asset(asset_id: int) -> dict[str, Any] | None:
|
||||
"""Renvoie aussi user_id — indispensable pour vérifier la PROPRIÉTÉ
|
||||
avant de servir/poser un asset (voir routes/assets/serve_user_asset.py,
|
||||
routes/scenes/scene_object_add.py), même esprit que l'isolation par
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
from .ensure_schema import ensure_users_schema
|
||||
from typing import Any
|
||||
|
||||
from .connection import connect
|
||||
from .ensure_schema import ensure_users_schema
|
||||
|
||||
|
||||
def get_user_by_email(email):
|
||||
def get_user_by_email(email: str | None) -> dict[str, Any] | None:
|
||||
ensure_users_schema()
|
||||
conn = connect()
|
||||
row = conn.execute("SELECT * FROM _users WHERE email = ?", ((email or "").strip().lower(),)).fetchone()
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
from .ensure_schema import ensure_users_schema
|
||||
from typing import Any
|
||||
|
||||
from .connection import connect
|
||||
from .ensure_schema import ensure_users_schema
|
||||
|
||||
|
||||
def get_user_by_id(user_id):
|
||||
def get_user_by_id(user_id: int | None) -> dict[str, Any] | None:
|
||||
ensure_users_schema()
|
||||
if not user_id:
|
||||
return None
|
||||
|
||||
@@ -1,8 +1,7 @@
|
||||
import re
|
||||
import struct
|
||||
|
||||
|
||||
def image_dimensions(content):
|
||||
def image_dimensions(content: bytes) -> tuple[int, int] | None:
|
||||
"""Largeur/hauteur RÉELLES (en pixels) d'une image, lues directement
|
||||
dans ses en-têtes — SANS dépendance externe (Pillow serait la solution
|
||||
habituelle, mais un simple parsing des formats courants suffit ici et
|
||||
@@ -32,7 +31,7 @@ def image_dimensions(content):
|
||||
return None
|
||||
|
||||
|
||||
def _webp_dimensions(content):
|
||||
def _webp_dimensions(content: bytes) -> tuple[int, int] | None:
|
||||
chunk = content[12:16]
|
||||
if chunk == b"VP8X" and len(content) >= 30:
|
||||
width = 1 + (content[24] | (content[25] << 8) | (content[26] << 16))
|
||||
@@ -50,7 +49,7 @@ def _webp_dimensions(content):
|
||||
return None
|
||||
|
||||
|
||||
def _jpeg_dimensions(content):
|
||||
def _jpeg_dimensions(content: bytes) -> tuple[int, int] | None:
|
||||
i = 2
|
||||
n = len(content)
|
||||
while i + 9 < n:
|
||||
@@ -63,9 +62,9 @@ def _jpeg_dimensions(content):
|
||||
continue
|
||||
if marker == 0xD9:
|
||||
break
|
||||
seg_len = struct.unpack(">H", content[i + 2:i + 4])[0]
|
||||
seg_len = struct.unpack(">H", content[i + 2 : i + 4])[0]
|
||||
if 0xC0 <= marker <= 0xCF and marker not in (0xC4, 0xC8, 0xCC):
|
||||
height, width = struct.unpack(">HH", content[i + 5:i + 9])
|
||||
height, width = struct.unpack(">HH", content[i + 5 : i + 9])
|
||||
return width, height
|
||||
i += 2 + seg_len
|
||||
return None
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
from .ensure_schema import ensure_users_schema
|
||||
from .connection import connect
|
||||
from .ensure_schema import ensure_users_schema
|
||||
|
||||
|
||||
def is_first_user():
|
||||
def is_first_user() -> bool:
|
||||
"""True s'il n'existe encore AUCUN compte — le tout premier compte créé
|
||||
devient automatiquement admin (voir create_user.py), pour ne jamais
|
||||
avoir besoin d'un mot de passe par défaut ou d'un script de bootstrap
|
||||
@@ -11,4 +11,4 @@ def is_first_user():
|
||||
conn = connect()
|
||||
count = conn.execute("SELECT COUNT(*) AS c FROM _users").fetchone()["c"]
|
||||
conn.close()
|
||||
return count == 0
|
||||
return bool(count == 0)
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
from typing import Any
|
||||
|
||||
from .connection import connect
|
||||
from .ensure_user_assets_schema import ensure_user_assets_schema
|
||||
|
||||
|
||||
def list_user_assets(user_id):
|
||||
""""Mes assets" d'un compte, plus récent en premier — voir
|
||||
def list_user_assets(user_id: int) -> list[dict[str, Any]]:
|
||||
""" "Mes assets" d'un compte, plus récent en premier — voir
|
||||
templates/scene_edit.html, bloc "Mes assets" du panneau d'ajout
|
||||
d'élément."""
|
||||
ensure_user_assets_schema()
|
||||
|
||||
@@ -7,11 +7,11 @@ from .connection import connect
|
||||
TOKEN_TTL_MINUTES = 60
|
||||
|
||||
|
||||
def _hash_token(token):
|
||||
def _hash_token(token: str) -> str:
|
||||
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def create_password_reset_token(user_id):
|
||||
def create_password_reset_token(user_id: int) -> str:
|
||||
"""Un seul jeton valide à la fois par utilisateur : en créer un
|
||||
nouveau invalide silencieusement tout jeu envoyé plus tôt (par
|
||||
exemple si l'utilisateur redemande un email parce que le premier
|
||||
@@ -29,7 +29,7 @@ def create_password_reset_token(user_id):
|
||||
return token
|
||||
|
||||
|
||||
def get_user_id_for_valid_token(token):
|
||||
def get_user_id_for_valid_token(token: str | None) -> int | None:
|
||||
if not token:
|
||||
return None
|
||||
conn = connect()
|
||||
@@ -42,10 +42,10 @@ def get_user_id_for_valid_token(token):
|
||||
return None
|
||||
if datetime.fromisoformat(row["expires_at"]) < datetime.now(timezone.utc):
|
||||
return None
|
||||
return row["user_id"]
|
||||
return int(row["user_id"])
|
||||
|
||||
|
||||
def consume_password_reset_token(token):
|
||||
def consume_password_reset_token(token: str) -> None:
|
||||
conn = connect()
|
||||
conn.execute(
|
||||
"UPDATE _password_reset_tokens SET used_at = CURRENT_TIMESTAMP WHERE token_hash = ?",
|
||||
|
||||
@@ -1,11 +1,12 @@
|
||||
import re
|
||||
from typing import Any, Callable
|
||||
|
||||
# Mêmes règles des deux côtés (ici pour le refus serveur, en JS dans
|
||||
# templates/auth/register.html pour le schéma visuel qui guide la saisie
|
||||
# en temps réel) — un mot de passe REFUSÉ côté serveur doit toujours
|
||||
# correspondre à une jauge/coche déjà rouge côté client, jamais une
|
||||
# surprise après coup.
|
||||
_CHECKS = [
|
||||
_CHECKS: list[tuple[str, str, Callable[[str], bool]]] = [
|
||||
("longueur", "Au moins 8 caractères", lambda p: len(p) >= 8),
|
||||
("majuscule", "Une majuscule", lambda p: re.search(r"[A-Z]", p) is not None),
|
||||
("minuscule", "Une minuscule", lambda p: re.search(r"[a-z]", p) is not None),
|
||||
@@ -19,7 +20,7 @@ _CHECKS = [
|
||||
MIN_SCORE_REQUIRED = 4
|
||||
|
||||
|
||||
def password_strength(password):
|
||||
def password_strength(password: str | None) -> dict[str, Any]:
|
||||
"""Renvoie {"score": 0-5, "checks": [{"key","label","ok"}, ...],
|
||||
"valid": bool} — jamais None, un mot de passe vide obtient juste un
|
||||
score de 0 (toutes les règles échouent), pas une erreur."""
|
||||
|
||||
+8
-6
@@ -3,24 +3,26 @@
|
||||
que soit l'étape attaquée) : 3 essais libres, puis un temps d'attente qui
|
||||
double à chaque échec supplémentaire (5 min, 10, 20, 40...), plafonné à
|
||||
1h. Remis à zéro dès une connexion réussie (voir reset_failed_attempts)."""
|
||||
|
||||
import math
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
_FREE_ATTEMPTS = 3
|
||||
_FIRST_LOCKOUT_MINUTES = 5
|
||||
_MAX_LOCKOUT_MINUTES = 60
|
||||
|
||||
|
||||
def lockout_minutes_for(failed_attempts):
|
||||
def lockout_minutes_for(failed_attempts: int) -> int:
|
||||
"""0 tant qu'on est dans les 3 essais libres ; sinon 5 * 2^n, plafonné
|
||||
à 60 — jamais négatif, jamais None."""
|
||||
if failed_attempts <= _FREE_ATTEMPTS:
|
||||
return 0
|
||||
exponent = failed_attempts - _FREE_ATTEMPTS - 1
|
||||
return min(_FIRST_LOCKOUT_MINUTES * (2 ** exponent), _MAX_LOCKOUT_MINUTES)
|
||||
return min(int(_FIRST_LOCKOUT_MINUTES * (2**exponent)), _MAX_LOCKOUT_MINUTES)
|
||||
|
||||
|
||||
def _parse(dt_str):
|
||||
def _parse(dt_str: str | None) -> datetime | None:
|
||||
if not dt_str:
|
||||
return None
|
||||
try:
|
||||
@@ -29,7 +31,7 @@ def _parse(dt_str):
|
||||
return None
|
||||
|
||||
|
||||
def lockout_message(seconds_remaining):
|
||||
def lockout_message(seconds_remaining: float) -> str:
|
||||
"""Message affiché tel quel dans le formulaire — arrondi à la minute
|
||||
supérieure (jamais "0 minute" alors qu'il en reste un peu, jamais un
|
||||
compte de secondes qui oblige à recharger sans arrêt pour voir si
|
||||
@@ -39,7 +41,7 @@ def lockout_message(seconds_remaining):
|
||||
return f"Trop de tentatives. Réessaie dans {minutes} {unit}."
|
||||
|
||||
|
||||
def seconds_locked_remaining(user):
|
||||
def seconds_locked_remaining(user: dict[str, Any] | None) -> int:
|
||||
"""> 0 si le compte est actuellement verrouillé (temps restant, en
|
||||
secondes, arrondi au-dessus pour ne jamais afficher "0 minute" alors
|
||||
qu'il en reste réellement un peu) ; 0 sinon."""
|
||||
|
||||
@@ -4,7 +4,7 @@ from .connection import connect
|
||||
from .rate_limit import lockout_minutes_for
|
||||
|
||||
|
||||
def record_failed_attempt(user_id):
|
||||
def record_failed_attempt(user_id: int) -> int:
|
||||
"""Incrémente le compteur d'échecs de CE compte et, une fois passé les
|
||||
3 essais libres, pose/allonge son verrouillage (voir rate_limit.py) —
|
||||
appelé sur un mot de passe OU un code 2FA incorrect, jamais sur un
|
||||
|
||||
@@ -7,11 +7,11 @@ from .connection import connect
|
||||
_CODE_COUNT = 10
|
||||
|
||||
|
||||
def _format_code(raw):
|
||||
return "-".join(raw[i:i + 4] for i in range(0, len(raw), 4))
|
||||
def _format_code(raw: str) -> str:
|
||||
return "-".join(raw[i : i + 4] for i in range(0, len(raw), 4))
|
||||
|
||||
|
||||
def generate_recovery_codes(user_id):
|
||||
def generate_recovery_codes(user_id: int) -> list[str]:
|
||||
"""(Re)génère les codes de récupération 2FA d'un utilisateur : un
|
||||
nouvel appel invalide tout jeu de codes précédent (un seul jeu valide
|
||||
à la fois, pour ne jamais avoir à deviner lesquels tiennent encore).
|
||||
@@ -21,7 +21,7 @@ def generate_recovery_codes(user_id):
|
||||
directement réutilisables."""
|
||||
conn = connect()
|
||||
conn.execute("DELETE FROM _recovery_codes WHERE user_id = ?", (user_id,))
|
||||
codes = []
|
||||
codes: list[str] = []
|
||||
for _ in range(_CODE_COUNT):
|
||||
code = _format_code(secrets.token_hex(6))
|
||||
codes.append(code)
|
||||
@@ -34,7 +34,7 @@ def generate_recovery_codes(user_id):
|
||||
return codes
|
||||
|
||||
|
||||
def verify_recovery_code(user_id, code):
|
||||
def verify_recovery_code(user_id: int, code: str | None) -> bool:
|
||||
"""Un code n'est utilisable qu'UNE SEULE FOIS (used_at) : consommé dès
|
||||
qu'il sert à une connexion réussie, pour qu'un code intercepté une
|
||||
fois (capture d'écran, historique du navigateur...) ne redonne pas un
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from .connection import connect
|
||||
|
||||
|
||||
def reset_failed_attempts(user_id):
|
||||
def reset_failed_attempts(user_id: int) -> None:
|
||||
"""Remet le compteur à zéro et lève tout verrouillage — appelé dès
|
||||
qu'une étape de connexion réussit (mot de passe validé ou code 2FA
|
||||
validé), pour ne jamais punir un utilisateur légitime qui s'est juste
|
||||
|
||||
@@ -5,7 +5,14 @@ from .create_user_asset import create_user_asset
|
||||
from .user_assets_dir import user_assets_dir
|
||||
|
||||
|
||||
def save_user_asset(user_id, content, ext, original_name=None, source="upload", scene_kind=None):
|
||||
def save_user_asset(
|
||||
user_id: int,
|
||||
content: bytes,
|
||||
ext: str,
|
||||
original_name: str | None = None,
|
||||
source: str = "upload",
|
||||
scene_kind: str | None = None,
|
||||
) -> tuple[int, str]:
|
||||
"""Écrit `content` (bytes) sur disque + crée la ligne "Mes assets" —
|
||||
factorisé pour être partagé par l'upload manuel
|
||||
(routes/assets/upload_user_asset.py) ET la génération IA
|
||||
|
||||
+3
-2
@@ -4,6 +4,7 @@ Python. Configuré uniquement par variables d'environnement (SMTP_HOST,
|
||||
SMTP_PORT, SMTP_USER, SMTP_PASSWORD, SMTP_FROM) : ce module ne connaît et
|
||||
ne stocke jamais le mot de passe SMTP en dur, à définir sur le poste/
|
||||
serveur qui fait tourner l'appli."""
|
||||
|
||||
import os
|
||||
import smtplib
|
||||
from email.mime.text import MIMEText
|
||||
@@ -19,7 +20,7 @@ class EmailNotConfiguredError(Exception):
|
||||
qu'il doit finir de configurer son SMTP."""
|
||||
|
||||
|
||||
def send_email(to_email, subject, body):
|
||||
def send_email(to_email: str, subject: str, body: str) -> None:
|
||||
host = os.environ.get("SMTP_HOST")
|
||||
port = int(os.environ.get("SMTP_PORT", "587"))
|
||||
user = os.environ.get("SMTP_USER")
|
||||
@@ -40,7 +41,7 @@ def send_email(to_email, subject, body):
|
||||
server.sendmail(sender, [to_email], msg.as_string())
|
||||
|
||||
|
||||
def send_password_reset_email(to_email, reset_url):
|
||||
def send_password_reset_email(to_email: str, reset_url: str) -> None:
|
||||
body = (
|
||||
"Une réinitialisation de mot de passe a été demandée pour ce compte "
|
||||
"Forge Engine.\n\n"
|
||||
|
||||
@@ -3,7 +3,7 @@ from werkzeug.security import generate_password_hash
|
||||
from .connection import connect
|
||||
|
||||
|
||||
def set_password(user_id, new_password):
|
||||
def set_password(user_id: int, new_password: str) -> None:
|
||||
"""Remet aussi le compteur anti-bruteforce à zéro (auth/rate_limit.py) :
|
||||
prouver son identité par email est une voie de récupération légitime,
|
||||
un compte verrouillé après trop d'échecs ne doit pas rester bloqué une
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from .connection import connect
|
||||
|
||||
|
||||
def set_project_slug(user_id, slug):
|
||||
def set_project_slug(user_id: int, slug: str) -> None:
|
||||
"""Enregistre le SEUL projet que ce compte possède (voir
|
||||
create_user.py — un compte "user" n'en aura jamais qu'un ; un compte
|
||||
"admin" n'a PAS besoin de cette colonne, il reste libre de créer
|
||||
|
||||
@@ -5,11 +5,11 @@ import qrcode
|
||||
import qrcode.image.svg
|
||||
|
||||
|
||||
def totp_provisioning_uri(secret, email):
|
||||
return pyotp.TOTP(secret).provisioning_uri(name=email, issuer_name="Forge Engine")
|
||||
def totp_provisioning_uri(secret: str, email: str) -> str:
|
||||
return str(pyotp.TOTP(secret).provisioning_uri(name=email, issuer_name="Forge Engine"))
|
||||
|
||||
|
||||
def totp_qrcode_svg(secret, email):
|
||||
def totp_qrcode_svg(secret: str, email: str) -> str:
|
||||
"""SVG (pas PNG) : la variante "image factory" par défaut de qrcode a
|
||||
besoin de Pillow pour produire un PNG — SvgPathImage, elle, est du pur
|
||||
Python, sans dépendance supplémentaire à installer juste pour un QR
|
||||
@@ -30,4 +30,4 @@ def totp_qrcode_svg(secret, email):
|
||||
buf = io.BytesIO()
|
||||
img.save(buf)
|
||||
svg = buf.getvalue().decode("utf-8")
|
||||
return svg[svg.index("<svg"):]
|
||||
return svg[svg.index("<svg") :]
|
||||
|
||||
@@ -9,7 +9,7 @@ class EmailUpdateError(Exception):
|
||||
valide, unicité), voir create_user.py."""
|
||||
|
||||
|
||||
def update_email(user_id, new_email):
|
||||
def update_email(user_id: int, new_email: str | None) -> str:
|
||||
new_email = (new_email or "").strip().lower()
|
||||
if not is_valid_email(new_email):
|
||||
raise EmailUpdateError("Adresse email invalide.")
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from .connection import connect
|
||||
|
||||
|
||||
def update_profile(user_id, nom, prenom):
|
||||
def update_profile(user_id: int, nom: str, prenom: str) -> None:
|
||||
conn = connect()
|
||||
conn.execute("UPDATE _users SET nom = ?, prenom = ? WHERE id = ?", (nom, prenom, user_id))
|
||||
conn.commit()
|
||||
|
||||
@@ -4,7 +4,7 @@ from .get_user_asset import get_user_asset
|
||||
from .user_asset_kind import user_asset_kind
|
||||
|
||||
|
||||
def update_user_asset_scene_kind(asset_id, user_id, scene_kind):
|
||||
def update_user_asset_scene_kind(asset_id: int, user_id: int, scene_kind: str | None) -> bool:
|
||||
"""Reclasse une image déjà importée entre "Fonds" et "Décors/objets"
|
||||
(voir create_user_asset.py) — SANS ré-upload (demande implicite,
|
||||
trouvée en diagnostiquant un "décalage" : une image importée AVANT
|
||||
|
||||
@@ -14,7 +14,7 @@ _AUDIO_EXTENSIONS = {".mp3", ".wav", ".ogg", ".m4a", ".aac", ".flac"}
|
||||
_VIDEO_EXTENSIONS = {".mp4", ".webm", ".mov", ".avi", ".ogv"}
|
||||
|
||||
|
||||
def user_asset_kind(filename):
|
||||
def user_asset_kind(filename: str) -> str:
|
||||
ext = os.path.splitext(filename)[1].lower()
|
||||
if ext in _IMAGE_EXTENSIONS:
|
||||
return "image"
|
||||
|
||||
@@ -3,7 +3,7 @@ import os
|
||||
from db.constants import USER_ASSETS_DIR
|
||||
|
||||
|
||||
def user_assets_dir(user_id):
|
||||
def user_assets_dir(user_id: int) -> str:
|
||||
"""Dossier disque des images d'un compte (voir ensure_user_assets_schema.py
|
||||
pour les métadonnées) — mirroir de db/game_dir.py, mais indexé par
|
||||
utilisateur plutôt que par slug de projet."""
|
||||
|
||||
@@ -9,7 +9,7 @@ from mutagen import File as MutagenFile
|
||||
MAX_AUDIO_SECONDS = 180
|
||||
|
||||
|
||||
def validate_audio_duration(content, max_seconds=MAX_AUDIO_SECONDS):
|
||||
def validate_audio_duration(content: bytes, max_seconds: int = MAX_AUDIO_SECONDS) -> str | None:
|
||||
"""Lit la durée d'un fichier audio (mp3/wav/ogg/m4a/... — mutagen
|
||||
détecte le format lui-même, pas besoin de le lui préciser) et renvoie
|
||||
un message d'erreur (français, prêt à afficher) si elle dépasse
|
||||
|
||||
@@ -9,7 +9,9 @@ from mutagen import File as MutagenFile
|
||||
MAX_VIDEO_SECONDS = 300
|
||||
|
||||
|
||||
def validate_video_duration(content, filename="video.mp4", max_seconds=MAX_VIDEO_SECONDS):
|
||||
def validate_video_duration(
|
||||
content: bytes, filename: str = "video.mp4", max_seconds: int = MAX_VIDEO_SECONDS
|
||||
) -> str | None:
|
||||
"""Lit la durée d'un fichier vidéo MP4 (seul format accepté à l'upload,
|
||||
voir routes/assets/upload_user_asset.py — mutagen ne sait pas lire
|
||||
fiablement la durée de webm/mov/avi, contrairement au conteneur MP4/
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
from typing import Any
|
||||
|
||||
from werkzeug.security import check_password_hash
|
||||
|
||||
|
||||
def verify_password(user, password):
|
||||
return bool(user) and check_password_hash(user["password_hash"], password or "")
|
||||
def verify_password(user: dict[str, Any] | None, password: str | None) -> bool:
|
||||
return user is not None and bool(check_password_hash(user["password_hash"], password or ""))
|
||||
|
||||
+2
-2
@@ -1,7 +1,7 @@
|
||||
import pyotp
|
||||
|
||||
|
||||
def verify_totp(secret, code):
|
||||
def verify_totp(secret: str, code: str | None) -> bool:
|
||||
"""valid_window=1 : tolère un léger décalage d'horloge entre le
|
||||
serveur et le téléphone (accepte aussi le code de la période
|
||||
précédente/suivante, ±30s) — sans quoi une horloge un peu désynchronisée
|
||||
@@ -9,4 +9,4 @@ def verify_totp(secret, code):
|
||||
code = (code or "").strip()
|
||||
if not code:
|
||||
return False
|
||||
return pyotp.TOTP(secret).verify(code, valid_window=1)
|
||||
return bool(pyotp.TOTP(secret).verify(code, valid_window=1))
|
||||
|
||||
Reference in New Issue
Block a user