Add automatic IP blacklisting after 2 illegal contributions

Persistence: PostgreSQL table ip_abuse (ip_hash SHA-256, rejection_count,
timestamps, last_idea_id, blacklisted_at, expires_at). No raw IP stored.

Logic:
- First illegal contribution: recorded, tolerated (benefit of the doubt)
- Second illegal contribution: 30-day block (IP_BLACKLIST_DAYS, configurable)
- Counter is cumulative — valid contributions do not reset it
- Block check fires before all other validations in both submit routes

Backend:
- database.py: check_ip_blacklist, record_ip_rejection, get_ip_blacklist,
  remove_ip_blacklist; ip_abuse table created in init_db()
- app.py: _get_ip_hash() helper; blacklist check + record in submit_idea()
  and submit_consultation_idea(); admin routes GET/DELETE /api/admin/ip-blacklist

Admin panel: new "Blacklist" tab showing active entries with hash, rejection
count, trigger idea id, dates; "Lever" button for manual removal

Docs:
- SECURITE_ANTI_ABUS.md: section 8 describing the mechanism, RGPD basis,
  limits, and IP_BLACKLIST_DAYS env var
- privacy-policy.tsx: IP row updated to reflect hash storage + legal basis

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-06-23 17:42:31 +02:00
parent 219292ea6c
commit 5c5152a387
5 changed files with 308 additions and 2 deletions
+81
View File
@@ -101,6 +101,19 @@ def init_db() -> None:
)
cur.execute("CREATE INDEX IF NOT EXISTS idx_ideas_consultation ON ideas(consultation_id)")
cur.execute("CREATE INDEX IF NOT EXISTS idx_synthesis_consultation ON synthesis(consultation_id)")
# Table anti-abus — blacklist IP (RGPD art. 6(1)(f) — intérêt légitime)
cur.execute("""
CREATE TABLE IF NOT EXISTS ip_abuse (
ip_hash TEXT PRIMARY KEY,
rejection_count INTEGER NOT NULL DEFAULT 1,
first_rejected_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_rejected_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_idea_id INTEGER,
blacklisted_at TIMESTAMPTZ,
expires_at TIMESTAMPTZ
)
""")
cur.execute("CREATE INDEX IF NOT EXISTS idx_ip_abuse_expires ON ip_abuse(expires_at)")
logger.info("Base de données initialisée.")
@@ -486,6 +499,74 @@ def get_consultation_contributions(
return rows, total
# ─── Anti-abus — Blacklist IP ─────────────────────────────────────────────────
def check_ip_blacklist(ip_hash: str) -> bool:
"""True si l'IP est blacklistée et que le blacklist n'a pas expiré."""
with db_cursor() as cur:
cur.execute(
"SELECT 1 FROM ip_abuse WHERE ip_hash = %s AND blacklisted_at IS NOT NULL AND expires_at > NOW()",
(ip_hash,),
)
return cur.fetchone() is not None
def record_ip_rejection(ip_hash: str, idea_id: int, blacklist_days: int) -> bool:
"""
Incrémente le compteur de rejets pour cette IP.
Déclenche le blacklist si rejection_count atteint 2.
Retourne True si le blacklist vient d'être activé.
"""
with db_cursor() as cur:
cur.execute(
"""
INSERT INTO ip_abuse (ip_hash, rejection_count, last_idea_id)
VALUES (%s, 1, %s)
ON CONFLICT (ip_hash) DO UPDATE
SET rejection_count = ip_abuse.rejection_count + 1,
last_rejected_at = NOW(),
last_idea_id = EXCLUDED.last_idea_id
RETURNING rejection_count, blacklisted_at
""",
(ip_hash, idea_id),
)
row = dict(cur.fetchone())
if row["rejection_count"] >= 2 and row["blacklisted_at"] is None:
cur.execute(
"""
UPDATE ip_abuse
SET blacklisted_at = NOW(),
expires_at = NOW() + (%s || ' days')::interval
WHERE ip_hash = %s
""",
(str(blacklist_days), ip_hash),
)
return True
return False
def get_ip_blacklist() -> list[dict]:
"""Liste des IPs blacklistées actives (non expirées), ordre anti-chronologique."""
with db_cursor() as cur:
cur.execute(
"""
SELECT ip_hash, rejection_count, first_rejected_at,
last_rejected_at, last_idea_id, blacklisted_at, expires_at
FROM ip_abuse
WHERE blacklisted_at IS NOT NULL AND expires_at > NOW()
ORDER BY blacklisted_at DESC
"""
)
return [dict(row) for row in cur.fetchall()]
def remove_ip_blacklist(ip_hash: str) -> bool:
"""Lève le blacklist d'une IP (supprime l'entrée entièrement)."""
with db_cursor() as cur:
cur.execute("DELETE FROM ip_abuse WHERE ip_hash = %s RETURNING ip_hash", (ip_hash,))
return cur.fetchone() is not None
def delete_consultation(consultation_id: int) -> bool:
"""Supprime une consultation et toutes ses données (idées, synthèse) dans la même transaction."""
with db_cursor() as cur: