Protections de Sécurité

Message entrant
       │
       ▼
┌─────────────────────┐
│  Liste blanche       │ ◄── Domaines/IPs de confiance
└──────┬──────────────┘
       │ non-whitelisté
       ▼
┌─────────────────────┐
│  Décompression safe  │ ◄── Protection contre les Decompression Bombs
└──────┬──────────────┘
       │
       ▼
┌─────────────────────┐
│  Résolution récursive│ ◄── Résolution des redirections (max 3 sauts)
└──────┬──────────────┘
       │
       ▼
┌─────────────────────┐
│  Scan multi-sources  │ ◄── PhishTank + GSB + VirusTotal
└──────┬──────────────┘
       │ menace détectée
       ▼
┌─────────────────────┐
│  Trust Score & Action│ ◄── Sanction adaptée au score
└─────────────────────┘

# Résolution avec protection contre les cycles et limite de profondeur
MAX_HOPS = 3
TIMEOUT_PER_HOP = 5  # secondes

async def resolve_redirects(url: str) -> tuple[str, list[str]]:
    visited = set()
    chain = [url]
    current = url

    for _ in range(MAX_HOPS):
        if current in visited:
            break  # Cycle détecté
        visited.add(current)

        response = await http_head(current, timeout=TIMEOUT_PER_HOP,
                                    allow_redirects=False)
        if response.status not in (301, 302, 303, 307, 308):
            break

        next_url = response.headers.get("Location")
        if not next_url:
            break

        current = normalize_url(next_url, base=current)
        chain.append(current)

    return current, chain

MAX_FILE_SIZE = 10 * 1024 * 1024   # 10 Mo maximum
MAX_DECOMPRESSED_SIZE = 50 * 1024 * 1024  # 50 Mo limite décompression
MAX_IMAGE_PIXELS = 4096 * 4096     # 16 MP limite de rendu

# Antiraid + scan concurrent limité
api_semaphore = asyncio.Semaphore(5)  # Max 5 requêtes parallèles

async def scan_url_with_ratelimit(url: str) -> ScanResult:
    async with api_semaphore:
        return await _scan_url_internal(url)

Probabilité de scan = max(0.1, 1.0 - (trust_score / 100) * 0.7)

# Exemple dans captcha.py
@tasks.loop(minutes=10)
async def cleanup_expired_sessions(self):
    now = asyncio.get_event_loop().time()
    expired = [
        session_id
        for session_id, session in self.active_sessions.items()
        if now - session.created_at > SESSION_TIMEOUT
    ]
    for session_id in expired:
        del self.active_sessions[session_id]

# utils/database.py
class Database:
    def __init__(self, pool):
        self._pool = pool  # Accès restreint via propriété

    @property
    def pool(self):
        return self._pool

    async def fetch_one(self, query: str, *args):
        async with self._pool.acquire() as conn:
            return await conn.fetchrow(query, *args)

# Chargement parallèle au lieu de séquentiel
trust_score, warnings, infractions = await asyncio.gather(
    db.fetch_trust_score(guild_id, user_id),
    db.fetch_warnings(guild_id, user_id),
    db.fetch_infractions(guild_id, user_id)
)

import secrets

def generate_captcha() -> tuple[str, int]:
    a = secrets.choice(range(1, 20))
    b = secrets.choice(range(1, 20))
    op = secrets.choice(['+', '-', '×'])
    # ...
    return question, correct_answer

Vérification avant ban/kick/mute :
  1. La cible est-elle un bot ? → Refus
  2. La cible est-elle le propriétaire du serveur ? → Refus
  3. Le rôle le plus haut de la cible ≥ rôle le plus haut du bot ? → Refus
  4. Le rôle le plus haut de la cible ≥ rôle le plus haut du modérateur ? → Refus
  5. Action autorisée ✓