In het verleden wij hebben elkaar ontmoet ELK-stapel, uit welke softwareproducten het bestaat. En de eerste taak waarmee een ingenieur wordt geconfronteerd wanneer hij met de ELK-stack werkt, is het verzenden van logboeken voor opslag in elasticsearch voor latere analyse. Dit is echter slechts lippendienst, elasticsearch slaat logs op in de vorm van documenten met bepaalde velden en waarden, wat betekent dat de ingenieur verschillende tools moet gebruiken om het bericht te ontleden dat vanaf de eindsystemen wordt verzonden. Dit kan op verschillende manieren: schrijf zelf een programma dat documenten aan de database toevoegt met behulp van de API, of gebruik kant-en-klare oplossingen. In deze cursus zullen we de oplossing bekijken Logstash, dat deel uitmaakt van de ELK-stack. We zullen bekijken hoe we logboeken van eindpuntsystemen naar Logstash kunnen sturen, en vervolgens zullen we een configuratiebestand opzetten om te parseren en om te leiden naar de Elasticsearch-database. Om dit te doen, nemen we logbestanden van de Check Point-firewall als binnenkomend systeem.
De cursus behandelt niet de installatie van de ELK-stack, aangezien er een groot aantal artikelen over dit onderwerp is; we zullen de configuratiecomponent bekijken.
Laten we een actieplan opstellen voor de Logstash-configuratie:
- Controleren of elasticsearch logbestanden accepteert (controle van de functionaliteit en openheid van de poort).
- We overwegen hoe we gebeurtenissen naar Logstash kunnen sturen, een methode kunnen kiezen en deze kunnen implementeren.
- We configureren Input in het Logstash-configuratiebestand.
- We configureren Output in het Logstash-configuratiebestand in debug-modus om te begrijpen hoe het logbericht eruit ziet.
- Filter instellen.
- De juiste uitvoer instellen in ElasticSearch.
- Logstash wordt gelanceerd.
- Het controleren van de logbestanden in Kibana.
Laten we elk punt in meer detail bekijken:
Controleren of elasticsearch logbestanden accepteert
Om dit te doen, kunt u de opdracht curl gebruiken om de toegang tot Elasticsearch te controleren vanaf het systeem waarop Logstash is geïmplementeerd. Als u authenticatie heeft geconfigureerd, dan dragen we de gebruiker/het wachtwoord ook over via curl, met vermelding van poort 9200 als u deze niet heeft gewijzigd. Als u een antwoord ontvangt dat lijkt op het onderstaande, dan is alles in orde.
[elastic@elasticsearch ~]$ curl -u <<user_name>> : <<password>> -sS -XGET "<<ip_address_elasticsearch>>:9200"
{
"name" : "elastic-1",
"cluster_name" : "project",
"cluster_uuid" : "sQzjTTuCR8q4ZO6DrEis0A",
"version" : {
"number" : "7.4.1",
"build_flavor" : "default",
"build_type" : "rpm",
"build_hash" : "fc0eeb6e2c25915d63d871d344e3d0b45ea0ea1e",
"build_date" : "2019-10-22T17:16:35.176724Z",
"build_snapshot" : false,
"lucene_version" : "8.2.0",
"minimum_wire_compatibility_version" : "6.8.0",
"minimum_index_compatibility_version" : "6.0.0-beta1"
},
"tagline" : "You Know, for Search"
}
[elastic@elasticsearch ~]$
Als het antwoord niet wordt ontvangen, kunnen er verschillende soorten fouten zijn: het elasticsearch-proces wordt niet uitgevoerd, de verkeerde poort is opgegeven of de poort wordt geblokkeerd door een firewall op de server waarop elasticsearch is geïnstalleerd.
Laten we eens kijken hoe u logboeken naar Logstash kunt sturen vanuit een controlepuntfirewall
Vanaf de Check Point-beheerserver kunt u logs naar Logstash sturen via syslog met behulp van het log_exporter-hulpprogramma. U kunt er hier meer over lezen , hier laten we alleen de opdracht achter die de stream maakt:
cp_log_export naam toevoegen check_point_syslog doelserver < > doelpoort 5555 protocol tcp-formaat generieke leesmodus semi-unified
< > - adres van de server waarop Logstash draait, doelpoort 5555 - poort waarnaar we logs zullen sturen, het verzenden van logs via tcp kan de server laden, dus in sommige gevallen is het correcter om udp te gebruiken.
INPUT instellen in het Logstash-configuratiebestand
Standaard bevindt het configuratiebestand zich in de map /etc/logstash/conf.d/. Het configuratiebestand bestaat uit 3 betekenisvolle delen: INPUT, FILTER, OUTPUT. IN INVOER we geven aan waar het systeem de logbestanden vandaan haalt FILTER parseer het logboek - stel in hoe het bericht in velden en waarden moet worden verdeeld OUTPUT we configureren de uitvoerstroom - waar de geparseerde logboeken naartoe worden verzonden.
Laten we eerst INPUT configureren en enkele van de typen bekijken die kunnen zijn: bestand, tcp en exe.
TCP:
input {
tcp {
port => 5555
host => “10.10.1.205”
type => "checkpoint"
mode => "server"
}
}
modus => "server"
Geeft aan dat Logstash verbindingen accepteert.
poort => 5555
gastheer => “10.10.1.205”
We accepteren verbindingen via IP-adres 10.10.1.205 (Logstash), poort 5555 - de poort moet worden toegestaan door het firewallbeleid.
typ => "controlepunt"
Wij markeren het document, erg handig als je meerdere inkomende verbindingen hebt. Vervolgens kunt u voor elke verbinding uw eigen filter schrijven met behulp van de logische if-constructie.
File:
input {
file {
path => "/var/log/openvas_report/*"
type => "openvas"
start_position => "beginning"
}
}
Beschrijving van instellingen:
pad => "/var/log/openvas_report/*"
Wij geven de directory aan waarin de bestanden gelezen moeten worden.
type => "openvaas"
Type evenement.
start_position => "begin"
Bij het wijzigen van een bestand leest het het gehele bestand; als u “end” instelt, wacht het systeem tot er nieuwe records aan het einde van het bestand verschijnen.
Directeur:
input {
exec {
command => "ls -alh"
interval => 30
}
}
Met behulp van deze invoer wordt een (enige!) shell-commando gelanceerd en de uitvoer ervan wordt omgezet in een logbericht.
commando => "ls -alh"
Het commando waarvan we de uitvoer willen.
interval => 30
Interval voor opdrachtaanroep in seconden.
Om logs van de firewall te ontvangen, registreren wij een filter tcp of udp, afhankelijk van hoe de logboeken naar Logstash worden verzonden.
We configureren Output in het Logstash-configuratiebestand in debug-modus om te begrijpen hoe het logbericht eruit ziet
Nadat we INPUT hebben geconfigureerd, moeten we begrijpen hoe het logbericht eruit zal zien en welke methoden moeten worden gebruikt om het logfilter (parser) te configureren.
Om dit te doen, zullen we een filter gebruiken dat het resultaat naar stdout uitvoert om het originele bericht te bekijken; het volledige configuratiebestand zal er op dit moment als volgt uitzien:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
output
{
if [type] == "checkpoint"
{
stdout { codec=> json }
}
}
Voer de opdracht uit om te controleren:
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
We zien het resultaat, de foto is klikbaar:
Als je het kopieert ziet het er zo uit:
action="Accept" conn_direction="Internal" contextnum="1" ifdir="outbound" ifname="bond1.101" logid="0" loguid="{0x5dfb8c13,0x5,0xfe0a0a0a,0xc0000000}" origin="10.10.10.254" originsicname="CN=ts-spb-cpgw-01,O=cp-spb-mgmt-01.tssolution.local.kncafb" sequencenum="8" time="1576766483" version="5" context_num="1" dst="10.10.10.10" dst_machine_name="[email protected]" layer_name="TSS-Standard Security" layer_name="TSS-Standard Application" layer_uuid="dae7f01c-4c98-4c3a-a643-bfbb8fcf40f0" layer_uuid="dbee3718-cf2f-4de0-8681-529cb75be9a6" match_id="8" match_id="33554431" parent_rule="0" parent_rule="0" rule_action="Accept" rule_action="Accept" rule_name="Implicit Cleanup" rule_uid="6dc2396f-9644-4546-8f32-95d98a3344e6" product="VPN-1 & FireWall-1" proto="17" s_port="37317" service="53" service_id="domain-udp" src="10.10.1.180" ","type":"qqqqq","host":"10.10.10.250","@version":"1","port":50620}{"@timestamp":"2019-12-19T14:50:12.153Z","message":"time="1576766483" action="Accept" conn_direction="Internal" contextnum="1" ifdir="outbound" ifname="bond1.101" logid="0" loguid="{0x5dfb8c13,
Als we naar deze berichten kijken, begrijpen we dat de logs er als volgt uitzien: veld = waarde of sleutel = waarde, wat betekent dat een filter met de naam kv geschikt is. Om voor elk specifiek geval het juiste filter te kiezen, is het een goed idee om er vertrouwd mee te raken in de technische documentatie, of om het aan een vriend te vragen.
Filter instellen
In de laatste fase hebben we kv geselecteerd, de configuratie van dit filter wordt hieronder weergegeven:
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
We selecteren het symbool waarmee we het veld en de waarde delen - “=”. Als we identieke vermeldingen in het logboek hebben, slaan we slechts één exemplaar in de database op, anders krijg je een array met identieke waarden, dat wil zeggen dat als we het bericht "foo = some foo=some" hebben, we alleen foo schrijven = sommige.
De juiste uitvoer instellen in ElasticSearch
Zodra Filter is geconfigureerd, kunt u logboeken naar de database uploaden elasticsearch:
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
Als het document is ondertekend met het controlepunttype, slaan we de gebeurtenis op in de elasticsearch-database, die standaard verbindingen accepteert op 10.10.1.200 op poort 9200. Elk document wordt opgeslagen in een specifieke index, in dit geval slaan we op in de index “checkpoint-” + huidige tijddatum. Elke index kan een specifieke set velden bevatten, of wordt automatisch aangemaakt wanneer een nieuw veld in een bericht verschijnt; veldinstellingen en hun type kunnen worden bekeken in toewijzingen.
Als u authenticatie hebt geconfigureerd (we zullen dit later bekijken), moeten de referenties voor het schrijven naar een specifieke index worden opgegeven. In dit voorbeeld is dat “tssolution” met het wachtwoord “cool”. U kunt onderscheid maken tussen gebruikersrechten om logboeken alleen naar een specifieke index te schrijven en niet meer.
Start Logstash.
Logstash-configuratiebestand:
input
{
tcp
{
port => 5555
type => "checkpoint"
mode => "server"
host => “10.10.1.205”
}
}
filter {
if [type] == "checkpoint"{
kv {
value_split => "="
allow_duplicate_values => false
}
}
}
output
{
if [type] == "checkpoint"
{
elasticsearch
{
hosts => ["10.10.1.200:9200"]
index => "checkpoint-%{+YYYY.MM.dd}"
user => "tssolution"
password => "cool"
}
}
}
We controleren het configuratiebestand op juistheid:
/usr/share/logstash/bin//logstash -f checkpoint.conf
Start het Logstash-proces:
sudo systemctl start logstash
We controleren of het proces is gestart:
sudo systemctl status logstash
Laten we controleren of de socket actief is:
netstat -nat |grep 5555
Het controleren van de logbestanden in Kibana.
Nadat alles draait, ga naar Kibana - Ontdek, zorg ervoor dat alles correct is geconfigureerd, de afbeelding is klikbaar!
Alle logbestanden zijn aanwezig en we kunnen alle velden en hun waarden zien!
Conclusie
We hebben gekeken hoe we een Logstash-configuratiebestand moesten schrijven, en als resultaat kregen we een parser van alle velden en waarden. Nu kunnen we werken met zoeken en plotten voor specifieke velden. Vervolgens gaan we in de cursus kijken naar visualisatie in Kibana en een eenvoudig dashboard maken. Het is de moeite waard te vermelden dat het Logstash-configuratiebestand in bepaalde situaties voortdurend moet worden bijgewerkt, bijvoorbeeld wanneer we de waarde van een veld willen vervangen van een getal naar een woord. In volgende artikelen zullen we dit voortdurend doen.
Dus blijf op de hoogte (, , , ), .
Bron: www.habr.com