Scripts Infrastructure

#!/usr/bin/env python3
"""
CrowdSec → Cloudflare IP Sync + AbuseIPDB Reporter + Recidivist Escalation
+ ModSecurity → CF Ban immédiat (2h) + Ban /24 automatique

1. Synchronise les bans actifs CrowdSec → Cloudflare IP Access Rules
2. Reporte les nouvelles IPs bannies (48h) → AbuseIPDB
3. Escalade les bans des récidivistes : 1er → CrowdSec gère | 2ème → 24h | 3ème+ → 7j
4. ModSecurity score ≥ 5 → ban CF 2h immédiat + report AbuseIPDB
5. Ban /24 automatique si 2+ IPs distinctes du même /24 en 7j → 24h

- Tourne toutes les 60 secondes
"""

import ipaddress
import json
import logging
import re
import subprocess
import time
import urllib.request
import urllib.error
import urllib.parse
from pathlib import Path
from datetime import datetime, timezone, timedelta

# ── Configuration ─────────────────────────────────────────────────
CF_API_TOKEN    = "VOTRE_CF_API_TOKEN"
CF_ZONE_ID      = "VOTRE_CF_ZONE_ID"
CS_API_KEY      = "VOTRE_CS_API_KEY"
ABUSEIPDB_KEY   = "VOTRE_ABUSEIPDB_KEY"
ABUSEIPDB_URL   = "https://api.abuseipdb.com/api/v2/report"
INTERVAL        = 60
NOTE_TAG        = "crowdsec-local-ban"
NOTE_TAG_MODSEC = "modsec-ban"
NOTE_TAG_CIDR   = "crowdsec-cidr-ban"
LOCAL_ORIGINS   = {"crowdsec", "cscli"}
DECISIONS_LOG   = Path("/var/log/crowdsec/decisions.log")
NGINX_ERROR_LOG = Path("/var/log/nginx/error.log")
CF_LOG_FILE     = Path("/var/log/crowdsec/cf-sync.log")
ABUSE_STATE     = Path("/var/log/crowdsec/abuseipdb-reported.json")
RECIDIV_STATE   = Path("/var/log/crowdsec/recidivists.json")
MODSEC_STATE    = Path("/var/log/crowdsec/modsec-banned.json")
CIDR_STATE      = Path("/var/log/crowdsec/cidr-banned.json")
LOOKBACK_HOURS  = 48
RECIDIV_WINDOW  = 7
CIDR_WINDOW     = 7
MODSEC_SCORE_MIN    = 5
MODSEC_BAN_SECS     = 7200
CIDR_BAN_DURATION   = "24h"
CIDR_THRESHOLD      = 2

RECIDIV_ESCALATION = {
    0: None,
    1: "24h",
}
RECIDIV_DEFAULT = "168h"

SCENARIO_CATEGORIES = {
    "http-sensitive-files":   "21",
    "http-probing":           "21",
    "http-scan":              "21,19",
    "http-bad-user-agent":    "21",
    "http-wordpress-scan":    "21",
    "http-crawl-non_statics": "21",
    "http-exploit":           "21",
    "vpatch-env-access":      "21",
    "vpatch-git-config":      "21",
    "ssh-bf":                 "22",
    "ssh-slow-bf":            "22",
    "ssh-time-based-bf":      "22",
    "ssh-refused-conn":       "22",
    "ssh-cve":                "22",
    "default":                "21",
}

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.FileHandler(CF_LOG_FILE), logging.StreamHandler()]
)
log = logging.getLogger(__name__)


def get_crowdsec_allowlist() -> set:
    try:
        result = subprocess.run(
            ["cscli", "allowlists", "inspect", "my_allowlist", "-o", "json"],
            capture_output=True, text=True, timeout=15
        )
        if result.returncode != 0:
            return set()
        data = json.loads(result.stdout)
        items = data.get("items", []) or []
        return {item.get("value", "") for item in items if item.get("value")}
    except Exception as e:
        log.warning("Erreur lecture allowlist CrowdSec: %s", e)
        return set()


def is_allowlisted(ip_str: str, cs_allowlist: set) -> bool:
    if ip_str in cs_allowlist:
        return True
    try:
        ip_obj = ipaddress.ip_address(ip_str)
        for entry in cs_allowlist:
            try:
                net = ipaddress.ip_network(entry, strict=False)
                if ip_obj in net:
                    return True
            except ValueError:
                pass
    except ValueError:
        pass
    return False


def cf_request(method: str, path: str, data=None) -> dict:
    url = f"https://api.cloudflare.com/client/v4{path}"
    body = json.dumps(data).encode() if data is not None else None
    req = urllib.request.Request(
        url, data=body, method=method,
        headers={
            "Authorization": f"Bearer {CF_API_TOKEN}",
            "Content-Type":  "application/json",
        }
    )
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            result = json.loads(resp.read().decode())
            if not result.get("success"):
                raise RuntimeError(f"CF API error: {result.get('errors')}")
            return result
    except urllib.error.HTTPError as e:
        body = e.read().decode()
        raise RuntimeError(f"HTTP {e.code} on {method} {path}: {body}") from e


def get_active_bans() -> set:
    """Bans actifs locaux (crowdsec + cscli) via cscli decisions list."""
    bans = set()
    for origin in LOCAL_ORIGINS:
        try:
            result = subprocess.run(
                ["cscli", "decisions", "list", "--origin", origin, "-o", "json"],
                capture_output=True, text=True, timeout=15
            )
            if result.returncode != 0 or not result.stdout.strip():
                continue
            alerts = json.loads(result.stdout)
            if not isinstance(alerts, list):
                continue
            for alert in alerts:
                for dec in alert.get("decisions") or []:
                    if (dec.get("type") == "ban"
                            and dec.get("scope", "").lower() == "ip"
                            and dec.get("value")):
                        bans.add(dec["value"])
        except Exception as e:
            log.warning("Erreur cscli decisions list --origin %s: %s", origin, e)
    return bans


def main():
    log.info("=== CrowdSec CF Sync démarré (interval=%ds) ===", INTERVAL)
    cs_allowlist = get_crowdsec_allowlist()
    log.info("Allowlist CrowdSec: %d entrées", len(cs_allowlist))

    while True:
        try:
            active_bans = get_active_bans()
            log.info("CrowdSec: %d bans actifs", len(active_bans))
            # ... (logique complète : sync CF, AbuseIPDB, récidivistes, ModSec, CIDR)
        except Exception as e:
            log.error("Erreur sync: %s", e, exc_info=True)
        time.sleep(INTERVAL)


if __name__ == "__main__":
    main()

(function () {
    'use strict';

    var COOKIE_NAME = 'arleo_cookie_consent';
    var COOKIE_DURATION = 365; // jours

    function getCookie(name) {
        var match = document.cookie.match(new RegExp('(^| )' + name + '=([^;]+)'));
        return match ? match[2] : null;
    }

    function setCookie(name, value, days) {
        var expires = new Date();
        expires.setTime(expires.getTime() + days * 24 * 60 * 60 * 1000);
        document.cookie = name + '=' + value
            + '; expires=' + expires.toUTCString()
            + '; path=/'
            + '; SameSite=Strict'
            + '; Secure';
    }

    function createBanner() {
        var banner = document.createElement('div');
        banner.id = 'arleo-cookie-banner';
        banner.setAttribute('role', 'dialog');
        banner.setAttribute('aria-live', 'polite');
        banner.setAttribute('aria-label', 'Consentement cookies');
        banner.innerHTML =
            '<div class="arleo-cookie-inner">' +
                '<p class="arleo-cookie-text">' +
                    'Ce site utilise des cookies de session nécessaires à son fonctionnement. ' +
                    '<a href="https://www.arleo.eu/privacy-policies" rel="noopener noreferrer">En savoir plus</a>' +
                '</p>' +
                '<button id="arleo-cookie-accept" aria-label="Accepter les cookies">OK</button>' +
            '</div>';

        // ... (CSS inline pour éviter une 2e requête HTTP)

        document.head.appendChild(style);
        document.body.appendChild(banner);

        document.getElementById('arleo-cookie-accept').addEventListener('click', function () {
            setCookie(COOKIE_NAME, 'accepted', COOKIE_DURATION);
            banner.style.transition = 'opacity 0.3s';
            banner.style.opacity = '0';
            setTimeout(function () {
                if (banner.parentNode) {
                    banner.parentNode.removeChild(banner);
                }
            }, 300);
        });
    }

    function init() {
        if (getCookie(COOKIE_NAME) === 'accepted') {
            return;
        }
        if (document.readyState === 'loading') {
            document.addEventListener('DOMContentLoaded', createBanner);
        } else {
            createBanner();
        }
    }

    init();
})();