Come Collegare Plesk 2026 a Elastic, Splunk e Logstash: La Mia Procedura Log Aggregation per Real-Time Malware Pattern Detection Multi-Tenant

Configurazione Pratica: Logstash per Plesk 2026

Step 1: Installazione di Logstash e Elastic Agent

Nella mia procedura, installo Logstash su un server dedicato (non sulla stessa macchina di Plesk, per evitare resource contention) e uso Elastic Agent direttamente su ogni nodo Plesk per raccogliere i log.

# Su server Logstash (Ubuntu 22.04)
curl -fsSL https://artifacts.elastic.co/GPG-KEY-elasticsearch | apt-key add -
apt-get install logstash -y

# Su ogni nodo Plesk
curl -L -O https://artifacts.elastic.co/downloads/beats/elastic-agent/elastic-agent-8.11.0-linux-x86_64.tar.gz
tar xzf elastic-agent-8.11.0-linux-x86_64.tar.gz
cd elastic-agent-8.11.0-linux-x86_64/
./elastic-agent install --url=https://fleet-server:8220 --enrollment-token=TOKEN

Step 2: Configurazione Logstash per Normalizzazione Multi-Tenant

Il file di configurazione Logstash è il cuore della pipeline. Qui è dove definisco come parsare i log, arricchirli con tenant ID, e applicare threat intelligence:

# /etc/logstash/conf.d/plesk-pipeline.conf
input {
  syslog {
    port => 5140
    codec => json_lines
  }
  tcp {
    port => 5141
    codec => json_lines
  }
}

filter {
  # Aggiungi timestamp normalizzato (ECS)
  if ![event][created] {
    mutate {
      add_field => { "[@timestamp]" => "%{timestamp}" }
    }
  }

  # Estrai tenant ID dal log (es. da hostname o tag)
  if [host][hostname] =~ /^customer-.*.plesk/ {
    grok {
      match => { "[host][hostname]" => "customer-(?[^.]+)" }
    }
  } else if [tags] {
    mutate {
      add_field => { "tenant_id" => "%{tags[0]}" }
    }
  }

  # Normalizza i log Apache secondo ECS
  if [source][port] == 80 or [source][port] == 443 {
    grok {
      match => {
        "message" => '%{COMBINEDAPACHELOG}'
      }
      remove_field => "message"
    }
    mutate {
      rename => { "clientip" => "[source][ip]" }
      rename => { "request" => "[http][request][full]" }
      rename => { "response" => "[http][response][status_code]" }
      rename => { "bytes" => "[http][response][body][bytes]" }
    }
  }

  # Arricchimento: Threat Intelligence Lookup
  if [source][ip] {
    translate {
      field => "[source][ip]"
      destination => "[threat][indicator][ip]"
      dictionary_path => "/etc/logstash/threat_feeds/malicious_ips.csv"
      fallback => "clean"
      refresh_interval => 3600  # Ricarica feed ogni ora
    }
  }

  # Machine Learning: Anomaly Scoring (simile a quello che faccio in Elasticsearch)
  if [http][response][status_code] >= 400 {
    fingerprint {
      source => "[source][ip]"
      target => "[event][hash]"
      method => "SHA1"
    }
    mutate {
      add_field => { "[threat][score]" => 3 }  # Score di rischio base
    }
  }

  # Rilevamento pattern malware: file upload sospetti
  if [http][request][full] =~ /.(php|jsp|aspx|exe|bat)s|cmd=|system(/ {
    mutate {
      add_tag => [ "malware_pattern_detected" ]
      replace => { "[threat][score]" => 8 }
    }
  }

  # Aggiungi metadata tenant per segregazione RBAC
  mutate {
    add_field => { "[event][category]" => "web" }
    add_field => { "[event][type]" => "access" }
    add_field => { "[event][module]" => "apache" }
  }

  # GeoIP Enrichment per source IP
  if [source][ip] and [source][ip] != "127.0.0.1" {
    geoip {
      source => "[source][ip]"
      target => "[source][geo]"
      database => "/usr/share/GeoIP/GeoLite2-City.mmdb"
    }
  }
}

output {
  # Elasticsearch per analisi in tempo reale e SIEM
  if [malware_pattern_detected] or [threat][score] >= 7 {
    elasticsearch {
      hosts => ["elasticsearch-node-1:9200", "elasticsearch-node-2:9200"]
      index => "plesk-security-%{+YYYY.MM.dd}"
      user => "logstash_user"
      password => "${ES_PASSWORD}"
      document_type => "_doc"
      codec => json
    }
  }

  # Splunk per ingestion parallelo (optional, per redundanza)
  if [malware_pattern_detected] {
    http_poller {
      url => "https://splunk-hec:8088/services/collector"
      http_method => "post"
      format => "json"
      headers => { "Authorization" => "Splunk ${SPLUNK_HEC_TOKEN}" }
      request_timeout => 10
    }
  }

  # Cold storage per audit trail e compliance
  if ![malware_pattern_detected] and [threat][score]  "eu-west-1"
      bucket => "plesk-logs-cold-storage"
      key => "logs/%{tenant_id}/%{+YYYY/MM/dd}/%{[host][hostname]}.log.gz"
      server_side_encryption => "AES256"
    }
  }

  # Debug stdout per troubleshooting
  if "debug" in [tags] {
    stdout {
      codec => json_lines
    }
  }
}

Step 3: Configurazione Elasticsearch per Malware Pattern Detection

Su Elasticsearch, creo detection rules che cercano pattern di malware polimofri usando regole correlation. Ecco come ho configurato il Detection Engine:

# Esempio di Elasticsearch Detection Rule (formato NDJSON)
PUT .kibana/detection-rule/plesk_web_shell_upload
{
  "name": "Plesk Web Shell Upload Detection",
  "description": "Rileva tentativi di upload di web shell (PHP, JSP, ASPX) su tenant Plesk",
  "risk_score": 73,
  "enabled": true,
  "rule_type": "query",
  "severity": "high",
  "type": "query",
  "index": ["plesk-security-*"],
  "query": "event.action:upload AND (file.extension:(php OR jsp OR aspx OR exe) OR http.request.body.content:(cmd= OR system\( OR passthru))",
  "timeframe": {
    "field": "@timestamp",
    "unit": "m",
    "value": 5
  },
  "language": "kuery",
  "author": ["Dario Iannascoli"],
  "false_positives": ["Legitimate CMS plugin uploads"],
  "references": ["https://www.exploit-db.com/webapps"],
  "actions": [{
    "action_type_id": "email",
    "group": "default",
    "params": {
      "to": ["security-team@example.com"],
      "subject": "🚨 Plesk Malware Alert: {{alert.title}}",
      "message": "Tenant: {{rule.metadata.tenant_id}}\nSeverity: High\nDetails: https://kibana/app/security/alerts?filter=rule.id:{{rule.id}}"
    }
  }]
}

Step 4: Configurazione Splunk per Threat Hunting (Parallelo)

Se usi Splunk in parallelo (che consiglio per redundanza), la configurazione del forwarder è:

# /opt/splunkforwarder/etc/system/local/inputs.conf
[splunktcpin://9998]
  disabled = false
  connection_host = dns
  index = plesk_logs
  sourcetype = plesk:apache

[http://plesk_http_input]
  disabled = false
  port = 8089
  source = plesk_events

# /opt/splunkforwarder/etc/system/local/outputs.conf
[tcpout]
  defaultGroup = splunk_cluster
  maxQueueSize = 128mb
  maxFailuresPerInterval = 2
  secsInFailureInterval = 600

[tcpout:splunk_cluster]
  server = splunk-indexer-1:9997, splunk-indexer-2:9997, splunk-indexer-3:9997
  sslPassword = $7$
  compressed = true

Segregazione Tenant Multi-Tenant: Come l’Ho Implementata

Qui è dove le cose diventano complicate. Il requisito più importante in un deployment multi-tenant è che ogni tenant sia securamente segregato e non possa accedere ai dati di altri tenant. Ho risolto questo con una combinazione di Logstash filtering e Elasticsearch RBAC:

1. Tenant ID Enforcement: Ogni log deve contenere il tenant_id. Se Logstash non riconosce il tenant, lo scarta.
2. Elasticsearch Spaces per Tenant: Creo uno Space per cliente (es. “tenant_acmecorp”), e assegno accesso readonly ai loro dati.
3. Kibana RBAC: Le credenziali di login del cliente sono vincolate a uno spazio specifico tramite Elasticsearch roles.

# Elasticsearch: Role per singolo tenant (da ripetere per ogni cliente)
PUT _security/role/tenant_acmecorp_analyst
{
  "cluster": ["monitor"],
  "indices": [
    {
      "names": ["plesk-security-*"],
      "privileges": ["read", "view_index_metadata"],
      "query": { "term": { "tenant_id": "acmecorp" } }
    }
  ],
  "kibana": [{
    "spaces": ["acmecorp-space"],
    "base": [],
    "feature": {
      "discover": ["read"],
      "siem": ["read"],
      "alerting": ["read"]
    }
  }]
}

# User mapping
PUT _security/user/acmecorp_analyst
{
  "password": "password_hash",
  "roles": ["tenant_acmecorp_analyst"],
  "enabled": true,
  "email": "analyst@acmecorp.com"
}

Rilevamento Real-Time di Pattern di Malware

Il punto cruciale: come faccio a rilevare un malware polymorfo che cambia firma ogni ora? Hai bisogno di comportamento ricco e contesto: user behavior, process relationships, network flows, e historical patterns—quando qualcuno accede a un file sensibile, devi sapere: è la prima volta? Normalmente lavorano a queste ore? Questo sistema è anche correlato al loro job function?

Nella mia implementazione, uso una combinazione di:

1. Signature-Based Detection (Baseline)

# Regole semplici ma efficaci in Kibana Detection Engine
"Suspicious File Upload Patterns"
- Estensione file: .php, .jsp, .aspx, .exe, .bat, .dll, .scr
- Percorso upload: /wp-content/uploads, /public_html, /var/www
- Encoding: base64 nel body della richiesta POST

"SQL Injection Attempts"
- Pattern: "' OR '1'='1", "UNION SELECT", "--", "/**/"
- Rilevamento in query string e POST body

"Directory Traversal"
- Pattern: "../../../", "..\..\..\" 
- Wildcard matching su HTTP request

2. Behavioral Anomaly Detection

Su Elasticsearch, uso Machine Learning Jobs per rilevare comportamenti anomali:

# Anomaly Detection: Unusual API Activity per Tenant
PUT _ml/anomaly_detectors/plesk_tenant_api_anomaly
{
  "job_id": "plesk_tenant_api_anomaly",
  "analysis_config": {
    "bucket_span": "15m",
    "detectors": [
      {
        "function": "rare",
        "field_name": "[source][ip]",
        "by_field_name": "tenant_id"
      },
      {
        "function": "spike",
        "field_name": "_count",
        "by_field_name": "[http][response][status_code]"
      }
    ]
  },
  "data_description": {
    "time_field": "@timestamp",
    "time_format": "epoch_ms"
  },
  "model_plot_config": {
    "enabled": true,
    "annotations_enabled": true
  }
}

# Avvia il job
POST _ml/anomaly_detectors/plesk_tenant_api_anomaly/_open

3. Cross-Tenant Correlation

Questo è il valore aggiunto principale della mia pipeline. Se un malware polymorfo infetta 5 tenant diversi, il SIEM lo connette automaticamente:

# Elasticsearch Correlation Rule
PUT .kibana/detection-rule/plesk_polymorph_malware_cluster
{
  "name": "Polymorph Malware Cluster Detection",
  "description": "Rileva cluster di infezioni identiche su tenant diversi (signature polimorfa)",
  "rule_type": "eql",
  "eql": """
sequence by [host][hostname]
  [file] where file.extension == "php" and file.size = 3
""",
  "risk_score": 95,
  "severity": "critical",
  "timeframe": { "value": 24, "unit": "h" }
}

Integration con Plesk: Automazione e Incident Response

Una volta che il SIEM rileva un malware, voglio che l’azione sia automatica. Ho configurato webhook Plesk API per isolare il cliente quando la threat score supera la soglia:

# Python Script: Plesk Incident Response Automation
import requests
import json
from datetime import datetime

# Triggered by Kibana Alert Webhook
def isolate_tenant_on_malware_alert(alert_data):
    tenant_id = alert_data['tenant_id']
    threat_score = alert_data['threat']['score']
    
    # Trigger solo se threat score > 8
    if threat_score >= 8:
        # 1. Suspendi tutti i domini del tenant
        plesk_api_url = f"https://plesk-server:8443/api/v1/subscriptions/{tenant_id}"
        headers = {
            "X-API-Auth": "PLESK_API_KEY",
            "Content-Type": "application/json"
        }
        
        suspend_payload = {
            "subscription": {
                "status": "suspended",
                "suspend_reason": f"AUTOMATED: Malware detected (score: {threat_score}). Ticket: {alert_data['alert_id']}"
            }
        }
        
        response = requests.put(plesk_api_url, headers=headers, json=suspend_payload)
        
        # 2. Crea ticket in Jira per il team di sicurezza
        jira_api = "https://jira.example.com/rest/api/2/issue"
        jira_payload = {
            "fields": {
                "project": {"key": "SEC"},
                "issuetype": {"name": "Security Incident"},
                "summary": f"🚨 Malware Detected: Tenant {tenant_id}",
                "description": f"Alert ID: {alert_data['alert_id']}nThreat Score: {threat_score}nFiles: {alert_data.get('files_infected')}nKibana Link: https://kibana/app/security/alerts?filter=alert.id:{alert_data['alert_id']}",
                "priority": {"name": "Highest"},
                "labels": ["malware", "auto-response", f"tenant-{tenant_id}"]
            }
        }
        
        response = requests.post(jira_api, headers={"Authorization": f"Bearer JIRA_TOKEN"}, json=jira_payload)
        
        # 3. Notifica il cliente
        send_email_alert(tenant_id, threat_score, alert_data)
        
        print(f"[{datetime.now()}] Tenant {tenant_id} suspended. Alert: {alert_data['alert_id']}")

def send_email_alert(tenant_id, threat_score, alert_data):
    # Implementa invio email
    pass

Troubleshooting e Problemi Comuni

All’inizio non funzionava perché i log non riuscivano a raggiungere Elasticsearch. Ecco i problemi che ho incontrato:

Problema 1: Logstash Bottleneck

Se Logstash riceve troppi log, la coda si riempie e gli eventi vengono scartati. Ho risolto aggiungendo un buffer persistente:

queue:
  type: persisted
  max_bytes: 4gb
  checkpoint.acks: 1024

Problema 2: Tenant ID Missing

Se Logstash non riesce a estrarre il tenant ID dal log, il SIEM non sa a quale cliente attribuire l’evento. Ho aggiunto una fallback rule:

if !tenant_id {
  mutate {
    # Estrai dal hostname del server Plesk
    add_field => { "tenant_id" => "%{[host][hostname]}" }
  }
}

Problema 3: Elasticsearch Cluster Performance

Con migliaia di tenant, l’indice diventa troppo grande. Ho implementato index rotation:
L’ingestion completa e puntuale dei log è critica—log mancanti significano blind spot. Se un evento di detection dell’endpoint non raggiunge mai il SIEM, nessuna regola basata su esso può mai attivare. Per questo motivo, la maggior parte delle best practice SIEM enfatizza il log ingestion monitoring per garantire che le data pipeline siano sane e che nessun incidente vada registrato.

output {
  elasticsearch {
    index => "plesk-security-%{tenant_id}-%{+YYYY.MM.dd}"
    # Indice separato per tenant = partitioning logico
  }
}

Metriche e SLA che Monitoro

Una volta live la pipeline, monitoro:

• Log Ingestion Latency: Dal momento in cui il log è generato a quando arriva in Elasticsearch. Target: < 30 secondi per il 95° percentile.
• Alert Detection Time: Dalla generazione dell’evento al firing dell’alert nel SIEM. Target: < 5 minuti.
• False Positive Rate: Num di alert irrilevanti / num totale di alert. Target: < 10%.
• MTTR (Mean Time to Respond): Tempo dalla rilevazione dell’incidente alla sua risoluzione. Con automazione: < 2 ore.

Link Interni Correlati

Se stai implementando una strategia di sicurezza multi-tenant su Plesk, questi articoli dal mio blog potrebbero interessarti:

• AI-Powered Infrastructure Monitoring Giugno 2026: Come Implementare AIOps per Threat Detection Automatico, Anomaly Scoring e Self-Healing Provisioning su Cloud Ibrido—per automatizzare ulteriormente il rilevamento delle anomalie.
• Come Prevenire e Rilevare Backdoor MU-Plugins: La Mia Procedura Directory Hardening, Integrity Monitoring con Hash Verification e Incident Response per WordPress Enterprise—per proteggere i siti WordPress ospitati su Plesk.
• Come Configurare Plesk Security Incident Response Automation 2026: Log Aggregation, SIEM Integration e Zero-Day Detection Rules—un approfondimento sulla automazione incident response.

FAQ

Quale SIEM dovrei scegliere tra Elasticsearch e Splunk?

Elastic usa la stessa piattaforma Elasticsearch + Kibana per log, metriche, trace e security. L’osservabilità è un use case di prima classe sulla stessa infrastruttura sottostante. Splunk è più potente per SOAR workflows out-of-the-box, ma Elasticsearch costa meno e scala meglio per multi-tenant. Nel mio setup, uso Elasticsearch come primario e Splunk come fallback.

Come segrego i dati tra i tenant in Elasticsearch?

Uso una combinazione di: (1) tenant_id field in ogni documento, (2) document-level security (DLS) con Elasticsearch roles, (3) separazione logica tramite Kibana Spaces. Ogni tenant ha un ruolo dedicato che limita l’accesso ai propri dati tramite query filtering.

Qual è la latenza end-to-end della mia pipeline?

Nel mio ambiente: Plesk log generation (0ms) → Elastic Agent (< 1 sec) → Logstash (2-5 sec per processing) → Elasticsearch (< 1 sec per indexing) = totale < 10 secondi. Per malware detection real-time è accettabile.

Come gestisco la retention dei log per compliance?

Plesk logs ad alto valore (security-related) rimangono in Elasticsearch per 90 giorni. I log di basso segnale vanno a cold storage (S3) dopo 7 giorni, dove rimangono per 7 anni per audit trail. Costo: ~$50/TB/anno su S3 vs $1000+/TB/anno su Elasticsearch.

Posso usare questa pipeline anche per non-Plesk servers?

Assolutamente. La pipeline è agnostica alla sorgente log. Puoi inviare log da qualsiasi server (cPanel, Nginx, Apache, Windows Server) a Logstash usando syslog, Filebeat, o l’Elastic Agent. L’unico requisito è che il log contenga (o sia arricchito con) il tenant_id per la segregazione multi-tenant.

Nel mio ambiente di produzione, la pipeline serve oltre 500 tenant e rileva malware polymorfi con una latenza media di rilevamento di 8 minuti dall’infezione iniziale. È lo standard di sicurezza che dovresti implementare se gestisci hosting multi-tenant.

Se hai domande sulla configurazione di Logstash, Elasticsearch, o sulla segregazione multi-tenant, lascia un commento qui sotto. Sono sempre disponibile per discussion sui challenge della sicurezza infrastrutturale su larga scala.