Cara mengirimkan log Alibaba Cloud ke SIEM - Simple Log Service

Kirim log dari Simple Log Service (SLS) ke platform Security Information and Event Management (SIEM) lokal, seperti Splunk atau QRadar. Hal ini memungkinkan Anda mengonsolidasikan log cloud dengan platform analitik keamanan lokal yang ada untuk pemantauan, audit, dan analisis ancaman terpadu.

Cara kerjanya

Aplikasi konsumen yang berada di dalam jaringan lokal Anda bertanggung jawab untuk mengambil data log. Aplikasi ini menggunakan kelompok konsumen SLS untuk menarik log secara real-time dan meneruskannya ke SIEM melalui protokol seperti Kolektor Peristiwa HTTP Splunk (HEC) atau Syslog melalui TCP/TLS.

Proses ini didasarkan pada arsitektur tarik, yang memberikan manfaat utama berikut:

Keamanan: Model tarik memastikan semua koneksi dimulai dari jaringan aman Anda. Tidak perlu membuka port masuk firewall, sehingga menjaga postur keamanan lokal Anda.
Throughput Tinggi dan Skalabilitas: Throughput tinggi dicapai melalui penskalaan horizontal. Anda dapat menjalankan beberapa instance konsumen secara bersamaan, dan kelompok konsumen secara otomatis menyeimbangkan beban kerja di seluruh semua instance aktif.
Reliabilitas: Kelompok konsumen menjamin pengiriman setidaknya sekali. Jika sebuah instance konsumen gagal, shard yang sedang diproses akan dialokasikan ulang ke instance sehat lainnya dalam kelompok. Konsumsi dilanjutkan dari checkpoint terakhir yang disimpan, mencegah kehilangan data.

Prasyarat

Izin: Buat Pengguna Resource Access Management (RAM), dan lampirkan kebijakan AliyunLogFullAccess kepada pengguna tersebut. Untuk informasi lebih lanjut, lihat dokumen terkait.
Persyaratan Jaringan: Mesin tempat program berjalan harus dapat mengakses titik akhir SLS dan berada di jaringan yang sama dengan SIEM.
- Untuk mendapatkan titik akhir:
  1. Masuk ke Konsol SLS. Di daftar proyek, klik proyek target.
  2. Klik ikon di sebelah kanan nama proyek untuk pergi ke halaman gambaran proyek.
  3. Di bagian Endpoint, lihat titik akhir publik. Titik akhir adalah https:// + titik akhir publik.
Persyaratan Lingkungan: Lingkungan runtime Python 3 dan SDK Python SLS.
1. Instal SDK Python SLS: pip install -U aliyun-log-python-sdk.
2. Verifikasi instalasi: pip show aliyun-log-python-sdk. Jika informasi berikut dikembalikan, instalasi berhasil.
```
Nama: aliyun-log-python-sdk
Versi: 0.9.12
Ringkasan: Aliyun log service Python client SDK
Halaman Utama: https://github.com/aliyun/aliyun-log-python-sdk
Penulis: Aliyun
```

Prosedur

Langkah 1: Persiapkan aplikasi

SLS menyediakan dua metode pengiriman:

Splunk HEC: HEC adalah mekanisme berbasis token yang memungkinkan Anda mengirim log dalam berbagai format data langsung ke Splunk melalui HTTP secara aman dan efisien.
Syslog: Saluran log umum yang kompatibel dengan sebagian besar SIEM dan mendukung format teks.

Splunk HEC

Untuk mengirimkan data log ke Splunk, konfigurasikan sync_data.py. Kode ini memiliki tiga bagian utama:

Metode main(): Titik masuk program utama.

Metode get_option(): Opsi konfigurasi konsumsi.

Opsi konfigurasi dasar: Termasuk pengaturan koneksi untuk SLS dan pengaturan kelompok konsumen.
Opsi lanjutan untuk kelompok konsumen: Parameter penyetelan performa. Jangan ubah kecuali diperlukan.
Parameter dan opsi terkait SIEM (Splunk).

Untuk melakukan pembersihan data selama proses pengiriman (seperti penyaringan baris, pemotongan kolom, atau normalisasi data), tambahkan aturan menggunakan kueri SPL. Contohnya:

# Pernyataan SPL
    query = "* | where instance_id in ('instance-1', 'instance-2')"
# Buat konsumen berdasarkan aturan. Dibandingkan dengan konsumsi normal, parameter query ditambahkan di akhir daftar parameter.
    option = LogHubConfig(endpoint, accessKeyId, accessKey, project, logstore, consumer_group, consumer_name,
                          cursor_position=CursorPosition.SPECIAL_TIMER_CURSOR,
                          cursor_start_time=cursor_start_time,
                          heartbeat_interval=heartbeat_interval,
                          data_fetch_interval=data_fetch_interval,
                          query=query)

SyncData(ConsumerProcessorBase): Berisi logika untuk mengambil data dari SLS dan mengirimkannya ke Splunk. Tinjau komentar dalam kode dengan cermat dan buat penyesuaian jika diperlukan.

Kode lengkapnya adalah sebagai berikut:

sync_data.py

# -*- coding: utf-8 -*-
import os
import logging
from logging.handlers import RotatingFileHandler
from aliyun.log.consumer import *
from aliyun.log.pulllog_response import PullLogResponse
from multiprocessing import current_process
import json
import socket
import requests

# Konfigurasikan file log rotasi untuk diagnosis dan pemecahan masalah.
root = logging.getLogger()
handler = RotatingFileHandler("{0}_{1}.log".format(os.path.basename(__file__), current_process().pid), maxBytes=100*1024*1024, backupCount=5)
handler.setFormatter(logging.Formatter(fmt='[%(asctime)s] - [%(threadName)s] - {%(module)s:%(funcName)s:%(lineno)d} %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S'))
root.setLevel(logging.INFO)
root.addHandler(handler)
root.addHandler(logging.StreamHandler())

logger = logging.getLogger(__name__)


class SyncData(ConsumerProcessorBase):
    """
    Konsumen mengonsumsi data dari Simple Log Service dan mengirimkannya ke Splunk.
    """
    def __init__(self, splunk_setting=None):
        
        """Inisialisasi dan verifikasi konektivitas ke Splunk."""
        super(SyncData, self).__init__()   # ingat untuk memanggil init dasar

        assert splunk_setting, ValueError("Anda perlu mengonfigurasi pengaturan target remote")
        assert isinstance(splunk_setting, dict), ValueError("Pengaturan harus berupa dict untuk mencakup alamat dan kredensial yang diperlukan.")

        self.option = splunk_setting
        self.timeout = self.option.get("timeout", 120)

        # Uji konektivitas ke Splunk.
        s = socket.socket()
        s.settimeout(self.timeout)
        s.connect((self.option["host"], self.option['port']))

        self.r = requests.session()
        self.r.max_redirects = 1
        self.r.verify = self.option.get("ssl_verify", True)
        self.r.headers['Authorization'] = "Splunk {}".format(self.option['token'])
        self.url = "{0}://{1}:{2}/services/collector".format("http" if not self.option.get('https') else "https", self.option['host'], self.option['port'])

        self.default_fields = {}
        if self.option.get("sourcetype"):
            self.default_fields['sourcetype'] = self.option.get("sourcetype")
        if self.option.get("source"):
            self.default_fields['source'] = self.option.get("source")
        if self.option.get("index"):
            self.default_fields['index'] = self.option.get("index")

    def process(self, log_groups, check_point_tracker):
        logs = PullLogResponse.loggroups_to_flattern_list(log_groups, time_as_str=True, decode_bytes=True)
        logger.info("Dapatkan data dari shard {0}, jumlah log: {1}".format(self.shard_id, len(logs)))
        for log in logs:
            # Ubah kode di sini: Ganti dengan kode pemrosesan sinkron untuk mengirim log ke ujung remote.
            # Format log adalah kamus. Contoh (Catatan: Semua string harus dienkoding Unicode):
            #    Python3: {"__time__": "12312312", "__topic__": "topic", "field1": "value1", "field2": "value2"}
            event = {}
            event.update(self.default_fields)
            event['time'] = log[u'__time__']
            del log['__time__']

            json_topic = {"actiontrail_audit_event": ["event"] }
            topic = log.get("__topic__", "")
            if topic in json_topic:
                try:
                    for field in json_topic[topic]:
                        log[field] = json.loads(log[field])
                except Exception as ex:
                    pass
            event['event'] = json.dumps(log)

            data = json.dumps(event, sort_keys=True)

            try:
                req = self.r.post(self.url, data=data, timeout=self.timeout)
                req.raise_for_status()
            except Exception as err:
                logger.debug("Gagal terhubung ke server Splunk remote ({0}). Pengecualian: {1}".format(self.url, err))
                raise err

                # Tambahkan beberapa percobaan ulang atau laporan sesuai kebutuhan.

        logger.info("Selesai mengirim data ke remote")

        self.save_checkpoint(check_point_tracker)


def get_option():
    ##########################
    # Item konfigurasi dasar
    ##########################

    # Muat parameter dan opsi Simple Log Service dari variabel lingkungan.
    accessKeyId = os.environ.get('SLS_AK_ID', '')
    accessKey = os.environ.get('SLS_AK_KEY', '')
    endpoint = os.environ.get('SLS_ENDPOINT', '')
    project = os.environ.get('SLS_PROJECT', '')
    logstore = os.environ.get('SLS_LOGSTORE', '')
    consumer_group = os.environ.get('SLS_CG', '')

    assert endpoint and accessKeyId and accessKey and project and logstore and consumer_group, \
        ValueError("endpoint/access_id/key/project/logstore/consumer_group/name tidak boleh kosong")

    ##########################
    # Opsi lanjutan untuk kelompok konsumen
    ##########################

    # Secara umum tidak disarankan untuk mengubah nama konsumen, terutama saat konsumsi bersamaan diperlukan.
    consumer_name = "{0}-{1}".format(consumer_group, current_process().pid)

    # Titik awal konsumsi. Parameter ini valid saat program dijalankan pertama kali. Jalankan berikutnya akan melanjutkan konsumsi dari checkpoint terakhir yang disimpan.
    # Gunakan "begin", "end", atau format waktu ISO tertentu.
    cursor_start_time = "2018-12-26 0:0:0"

    # Interval detak jantung. Jika server tidak menerima laporan detak jantung untuk shard tertentu dalam dua kali interval, server menganggap konsumen yang sesuai offline dan menugaskan ulang tugas.
    # Saat lingkungan jaringan buruk, tidak disarankan untuk mengatur interval pendek.
    heartbeat_interval = 20

    # Interval maksimum untuk konsumsi data. Jika data dihasilkan dengan cepat, Anda tidak perlu menyesuaikan parameter ini.
    data_fetch_interval = 1
    
    # Bangun kelompok konsumen dan konsumen.
    option = LogHubConfig(endpoint, accessKeyId, accessKey, project, logstore, consumer_group, consumer_name,
                          cursor_position=CursorPosition.SPECIAL_TIMER_CURSOR,
                          cursor_start_time=cursor_start_time,
                          heartbeat_interval=heartbeat_interval,
                          data_fetch_interval=data_fetch_interval)
"""
    Saat membangun konsumen berdasarkan aturan, gunakan kode berikut:
    # Pernyataan SPL kustom
    query = "* | where instance_id in ('instance-1', 'instance-2')"
    # Bangun konsumsi berdasarkan aturan. Dibandingkan dengan konsumsi normal, parameter query ditambahkan di akhir daftar parameter.
    option = LogHubConfig(endpoint, accessKeyId, accessKey, project, logstore, consumer_group, consumer_name,
                          cursor_position=CursorPosition.SPECIAL_TIMER_CURSOR,
                          cursor_start_time=cursor_start_time,
                          heartbeat_interval=heartbeat_interval,
                          data_fetch_interval=data_fetch_interval,
                          query=query)
    """

    # Opsi Splunk
    settings = {
                "host": "1.2.3.4",
                "port": 80,
                "token": "a0*****123",
                'https': False,              # Opsional, bool
                'timeout': 120,             # Opsional, int
                'ssl_verify': True,         # Opsional, bool
                "sourcetype": "",            # Opsional, sourcetype
                "index": "",                # Opsional, index
                "source": "",               # Opsional, source
            }

    return  option, settings

# Titik masuk program utama  
def main():
    option, settings = get_option()

    logger.info("*** mulai mengonsumsi data...")
    worker = ConsumerWorker(SyncData, option, args=(settings,) )
    worker.start(join=True)

if __name__ == '__main__':
    main()

Syslog

Syslog mendefinisikan spesifikasi format log terutama berdasarkan RFC5424 dan RFC3164. Kami merekomendasikan menggunakan protokol RFC 5424. Meskipun TCP dan UDP dapat mentransmisikan Syslog, TCP menyediakan transmisi data yang lebih andal daripada UDP. Protokol RFC 5424 juga mendefinisikan lapisan transport aman menggunakan TLS. Jika SIEM Anda mendukung saluran TCP atau TLS untuk Syslog, kami merekomendasikan menggunakan saluran tersebut.

Untuk mengirimkan data log ke SIEM menggunakan Syslog, konfigurasikan sync_data.py. Kode ini memiliki tiga bagian utama:

Metode main(): Titik masuk program utama.
Metode get_monitor_option(): Opsi konfigurasi konsumsi.
- Opsi konfigurasi dasar: Termasuk pengaturan koneksi untuk SLS dan pengaturan kelompok konsumen.
- Opsi lanjutan untuk kelompok konsumen: Parameter penyetelan performa. Jangan ubah kecuali diperlukan.
- Parameter dan opsi terkait server Syslog SIEM.
  - Fasilitas Syslog: Komponen program. Contoh menggunakan syslogclient.FAC_USER sebagai default.
  - Tingkat keparahan Syslog: Tingkat log. Atur tingkat log untuk konten tertentu sesuai kebutuhan. Di sini, syslogclient.SEV_INFO dipilih.
  - Jika SIEM mendukung saluran Syslog berbasis TCP atau TLS, atur proto ke TLS dan konfigurasikan Sertifikat SSL yang benar.
SyncData(ConsumerProcessorBase): Berisi logika tentang cara mengambil data dari SLS dan mengirimkannya ke server Syslog SIEM. Baca komentar dalam kode dengan cermat dan buat penyesuaian jika diperlukan.

Kode lengkapnya adalah sebagai berikut:

sync_data.py

# -*- coding: utf-8 -*-

import os
import logging
from logging.handlers import RotatingFileHandler
from aliyun.log.consumer import *
from aliyun.log.pulllog_response import PullLogResponse
from multiprocessing import current_process
import aliyun.log.ext.syslogclient as syslogclient
from aliyun.log.ext.syslogclient import SyslogClientRFC5424 as SyslogClient
import six
from datetime import datetime

# Konfigurasikan file log rotasi untuk diagnosis dan pemecahan masalah.
root = logging.getLogger()
handler = RotatingFileHandler("{0}_{1}.log".format(os.path.basename(__file__), current_process().pid), maxBytes=100*1024*1024, backupCount=5)
handler.setFormatter(logging.Formatter(fmt='[%(asctime)s] - [%(threadName)s] - {%(module)s:%(funcName)s:%(lineno)d} %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S'))
root.setLevel(logging.INFO)
root.addHandler(handler)
root.addHandler(logging.StreamHandler())

logger = logging.getLogger(__name__)


class SyncData(ConsumerProcessorBase):
    """
   Konsumen mengonsumsi data dari Simple Log Service dan mengirimkannya ke server Syslog.
    """
    def __init__(self, target_setting=None):
        """
        Inisialisasi dan verifikasi konektivitas ke server Syslog.
        """

        super(SyncData, self).__init__()   # ingat untuk memanggil init dasar

        assert target_setting, ValueError("Anda perlu mengonfigurasi pengaturan target remote")
        assert isinstance(target_setting, dict), ValueError("Pengaturan harus berupa dict untuk mencakup alamat dan kredensial yang diperlukan.")

        self.option = target_setting
        self.protocol = self.option['protocol']
        self.timeout = int(self.option.get('timeout', 120))
        self.sep = self.option.get('sep', "||")
        self.host = self.option["host"]
        self.port = int(self.option.get('port', 514))
        self.cert_path=self.option.get('cert_path', None)

        # Uji konektivitas. 
        with SyslogClient(self.host, self.port, proto=self.protocol, timeout=self.timeout, cert_path=self.cert_path) as client:
            pass

    def process(self, log_groups, check_point_tracker):
        logs = PullLogResponse.loggroups_to_flattern_list(log_groups, time_as_str=True, decode_bytes=True)
        logger.info("Dapatkan data dari shard {0}, jumlah log: {1}".format(self.shard_id, len(logs)))

        try:
            with SyslogClient(self.host, self.port, proto=self.protocol, timeout=self.timeout, cert_path=self.cert_path) as client:
                for log in logs:
                    # Tempatkan kode sinkron untuk mengirim log ke ujung remote di sini.
                    # Format log adalah kamus. Contoh (Catatan: Semua string harus dienkoding Unicode):
                    #    Python2: {"__time__": "12312312", "__topic__": "topic", u"field1": u"value1", u"field2": u"value2"}
                    #    Python3: {"__time__": "12312312", "__topic__": "topic", "field1": "value1", "field2": "value2"}
                
                    timestamp = datetime.fromtimestamp(int(log[u'__time__']))
                    del log['__time__']

                    io = six.StringIO()
                    first = True
                    # Ubah format sesuai kebutuhan. Di sini, Key=Value digunakan untuk transmisi, dan garis vertikal ganda (||) digunakan sebagai pemisah default.
                    for k, v in six.iteritems(log):
                        io.write("{0}{1}={2}".format(self.sep, k, v))

                    data = io.getvalue()
                    # Ubah fasilitas atau tingkat keparahan sesuai kebutuhan.
                    client.log(data, facility=self.option.get("facility", None), severity=self.option.get("severity", None), timestamp=timestamp, program=self.option.get("tag", None), hostname=self.option.get("hostname", None))

        except Exception as err:
            logger.debug("Gagal terhubung ke server syslog remote ({0}). Pengecualian: {1}".format(self.option, err))
            # Anda perlu menambahkan kode penanganan kesalahan, seperti percobaan ulang atau notifikasi. 
            raise err

        logger.info("Selesai mengirim data ke remote")

        self.save_checkpoint(check_point_tracker)


def get_monitor_option():
    ##########################
    # Item konfigurasi dasar
    ##########################

    # Muat parameter dan opsi Simple Log Service dari variabel lingkungan.
    endpoint = os.environ.get('SLS_ENDPOINT', '')
    accessKeyId = os.environ.get('SLS_AK_ID', '')
    accessKey = os.environ.get('SLS_AK_KEY', '')
    project = os.environ.get('SLS_PROJECT', '')
    logstore = os.environ.get('SLS_LOGSTORE', '')
    consumer_group = os.environ.get('SLS_CG', '')

    assert endpoint and accessKeyId and accessKey and project and logstore and consumer_group, \
        ValueError("endpoint/access_id/key/project/logstore/consumer_group/name tidak boleh kosong")

    ##########################
    # Opsi lanjutan untuk kelompok konsumen
    ##########################

    # Secara umum tidak disarankan untuk mengubah nama konsumen, terutama saat konsumsi bersamaan diperlukan.
    consumer_name = "{0}-{1}".format(consumer_group, current_process().pid)

    # Titik awal konsumsi. Parameter ini valid saat program dijalankan pertama kali. Jalankan berikutnya akan melanjutkan konsumsi dari checkpoint terakhir yang disimpan.
    # Gunakan "begin", "end", atau format waktu ISO tertentu.
    cursor_start_time = "2019-1-1 0:0:0+8:00"

    # Interval detak jantung. Jika server tidak menerima laporan detak jantung untuk shard tertentu dalam dua kali interval, server menganggap konsumen yang sesuai offline dan menugaskan ulang tugas.
    # Saat lingkungan jaringan buruk, tidak disarankan untuk mengatur interval pendek.
    heartbeat_interval = 20

    # Interval maksimum untuk konsumsi data. Jika data dihasilkan dengan cepat, Anda tidak perlu menyesuaikan parameter ini.
    data_fetch_interval = 1

    # Bangun kelompok konsumen dan konsumen.
    option = LogHubConfig(endpoint, accessKeyId, accessKey, project, logstore, consumer_group, consumer_name,
                          cursor_position=CursorPosition.SPECIAL_TIMER_CURSOR,
                          cursor_start_time=cursor_start_time,
                          heartbeat_interval=heartbeat_interval,
                          data_fetch_interval=data_fetch_interval)

    # Parameter dan opsi terkait server Syslog
    settings = {
                "host": "1.2.3.4", # Diperlukan
                "port": 514,       # Diperlukan, port
                "protocol": "tcp", # Diperlukan, TCP, UDP, atau TLS (Python 3 saja)
                "sep": "||",      # Diperlukan, pemisah untuk pasangan key=value. Di sini, garis vertikal ganda (||) digunakan.
                "cert_path": None,  # Opsional, lokasi sertifikat TLS
                "timeout": 120,   # Opsional, periode timeout, default adalah 120 detik
                "facility": syslogclient.FAC_USER,  # Opsional, lihat nilai syslogclient.FAC_* lainnya
                "severity": syslogclient.SEV_INFO,  # Opsional, lihat nilai syslogclient.SEV_* lainnya
                "hostname": None, # Opsional, nama mesin, nama mesin lokal digunakan secara default
                "tag": None # Opsional, tag, default adalah tanda hubung (-)
    }

    return option, settings

# Titik masuk program utama
def main():
    option, settings = get_monitor_option()

    logger.info("*** mulai mengonsumsi data...")
    worker = ConsumerWorker(SyncData, option, args=(settings,) )
    worker.start(join=True)


if __name__ == '__main__':
    main()

Langkah 2: Konfigurasikan variabel lingkungan

Setelah mengonfigurasi aplikasi, atur variabel lingkungan sistem seperti yang dijelaskan dalam tabel berikut.

Variabel lingkungan	Nilai	Contoh
SLS_ENDPOINT	Masuk ke Konsol Simple Log Service. Di daftar proyek, klik proyek target. Klik ikon di sebelah kanan nama proyek untuk pergi ke halaman gambaran proyek. Di bagian Endpoint, salin titik akhir publik dan gabungkan sebagai `https://` + titik akhir publik. Jika titik akhir diawali dengan `https://`, seperti `https://cn-beijing.log.aliyuncs.com`, program secara otomatis menggunakan HTTPS untuk mengenkripsi koneksi dengan SLS. Sertifikat server `*.aliyuncs.com` diterbitkan oleh GlobalSign dan secara default dipercaya oleh sebagian besar sistem. Jika sistem Anda tidak mempercayai sertifikat ini, unduh dan instal melalui Pemasangan Sertifikat.	`https://cn-beijing.log.aliyuncs.com`
SLS_PROJECT	Di Konsol SLS, salin nama proyek target.	my-sls-project-one
SLS_LOGSTORE	Di Konsol SLS, salin nama logstore target.	my-sls-logstore-a1
SLS_AK_ID	Gunakan ID AccessKey dari pengguna RAM. Penting Pasangan AccessKey akun Alibaba Cloud memberikan akses API penuh. Untuk keamanan yang lebih baik, selalu gunakan pasangan AccessKey pengguna RAM untuk panggilan API atau operasi pemeliharaan rutin. Untuk mencegah kebocoran kredensial, hindari menuliskan ID AccessKey dan Rahasia AccessKey secara langsung dalam kode Anda.	L***ky
SLS_AK_KEY	Gunakan Rahasia AccessKey dari pengguna RAM.	x***Xl
SLS_CG	Nama kelompok konsumen. Jika kelompok yang ditentukan tidak ada, aplikasi akan membuatnya secara otomatis.	syc_data

Langkah 3: Mulai dan verifikasi

Mulai beberapa konsumen untuk konsumsi bersamaan. Jumlah maksimum konsumen sama dengan jumlah total shard.

# Mulai proses konsumen pertama
nohup python3 sync_data.py &
# Mulai proses konsumen kedua
nohup python3 sync_data.py &

Periksa status kelompok konsumen di Konsol SLS.
1. Di daftar proyek, klik proyek target. Pada tab Log Storage > Logstores, klik ikon di sebelah logstore target, lalu klik ikon di sebelah Data Consumption.
2. Di daftar kelompok konsumen, klik yang Anda inginkan. Pada halaman Consumer Group Status, lihat klien dan waktu konsumsi data untuk setiap shard.

FAQ

Kesalahan ConsumerGroupQuotaExceed terjadi

Kesalahan ini menunjukkan bahwa batas telah terlampaui. Sebuah logstore dapat memiliki maksimal 30 kelompok konsumen. Hapus kelompok konsumen yang tidak digunakan di Konsol SLS.