2. 탄력적 스택: 보안 로그 분석. 로그스태시

과거에 기사 우리는 만났다 ELK 스택, 어떤 소프트웨어 제품으로 구성되어 있는지. 그리고 엔지니어가 ELK 스택으로 작업할 때 직면하는 첫 번째 작업은 후속 분석을 위해 Elasticsearch에 저장할 로그를 보내는 것입니다. 그러나 이것은 단지 립서비스에 불과합니다. Elasticsearch는 특정 필드와 값을 포함하는 문서 형식으로 로그를 저장합니다. 이는 엔지니어가 최종 시스템에서 전송된 메시지를 구문 분석하기 위해 다양한 도구를 사용해야 함을 의미합니다. 이는 여러 가지 방법으로 수행할 수 있습니다. API를 사용하여 데이터베이스에 문서를 추가하는 프로그램을 직접 작성하거나 기성 솔루션을 사용하십시오. 이 과정에서 우리는 해결책을 고려할 것입니다 로그 대쉬, 이는 ELK 스택의 일부입니다. 엔드포인트 시스템에서 Logstash로 로그를 보내는 방법을 살펴본 다음, Elasticsearch 데이터베이스를 구문 분석하고 리디렉션하기 위한 구성 파일을 설정하겠습니다. 이를 위해 Check Point 방화벽의 로그를 수신 시스템으로 사용합니다.

이 주제에 대한 기사가 너무 많기 때문에 ELK 스택 설치에 대해서는 다루지 않습니다. 구성 구성 요소를 고려해 보겠습니다.

Logstash 구성을 위한 실행 계획을 작성해 보겠습니다.

1. Elasticsearch가 로그를 허용하는지 확인합니다(포트의 기능 및 개방성 확인).
2. Logstash에 이벤트를 전송하고 방법을 선택하고 구현하는 방법을 고려합니다.
3. Logstash 구성 파일에서 입력을 구성합니다.
4. 로그 메시지의 모양을 이해하기 위해 디버그 모드에서 Logstash 구성 파일의 출력을 구성합니다.
5. 필터를 설정 중입니다.
6. ElasticSearch에서 올바른 출력을 설정합니다.
7. 로그스태시가 출시됩니다.
8. Kibana에서 로그를 확인합니다.

각 사항을 더 자세히 살펴보겠습니다.

Elasticsearch가 로그를 허용하는지 확인

이를 위해, 컬 명령을 사용하여 Logstash가 배포된 시스템에서 Elasticsearch에 대한 액세스를 확인할 수 있습니다. 인증이 구성되어 있으면 컬을 통해 사용자/비밀번호도 전송하며, 포트를 변경하지 않은 경우 포트 9200을 지정합니다. 아래와 비슷한 응답이 나타나면 모든 것이 정상인 것입니다.

[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 ~]$

응답이 수신되지 않으면 여러 유형의 오류가 있을 수 있습니다. Elasticsearch 프로세스가 실행되고 있지 않거나, 잘못된 포트가 지정되었거나, Elasticsearch가 설치된 서버의 방화벽에 의해 포트가 차단되었습니다.

체크포인트 방화벽에서 Logstash로 로그를 보내는 방법을 살펴보겠습니다.

Check Point 관리 서버에서 log_exporter 유틸리티를 사용하여 syslog를 통해 Logstash에 로그를 보낼 수 있습니다. 자세한 내용은 여기에서 확인할 수 있습니다. 기사, 여기서는 스트림을 생성하는 명령만 남겨 두겠습니다.

cp_log_export 이름 추가 check_point_syslog 대상 서버 < > 대상 포트 5555 프로토콜 tcp 형식 일반 읽기 모드 반통합

< > - Logstash가 실행되는 서버의 주소, target-port 5555 - 로그를 보낼 포트, tcp를 통해 로그를 보내면 서버를 로드할 수 있으므로 어떤 경우에는 udp를 사용하는 것이 더 정확합니다.

Logstash 구성 파일에서 INPUT 설정

기본적으로 구성 파일은 /etc/logstash/conf.d/ 디렉터리에 있습니다. 구성 파일은 INPUT, FILTER, OUTPUT의 세 가지 의미 있는 부분으로 구성됩니다. 안에 입력 시스템이 로그를 가져올 위치를 나타냅니다. FILTER 로그 분석 - 메시지를 필드와 값으로 나누는 방법을 설정합니다. 출력 구문 분석된 로그가 전송될 출력 스트림을 구성합니다.

먼저 INPUT을 구성하고 파일, tcp 및 exe와 같은 몇 가지 유형을 고려해 보겠습니다.

TCP:

input {
tcp {
    port => 5555
    host => “10.10.1.205”
    type => "checkpoint"
    mode => "server"
}
}

모드 => "서버"
Logstash가 연결을 수락하고 있음을 나타냅니다.

포트 => 5555
호스트 => “10.10.1.205”
IP 주소 10.10.1.205(Logstash), 포트 5555를 통한 연결을 허용합니다. 포트는 방화벽 정책에 의해 허용되어야 합니다.

유형 => "체크포인트"
들어오는 연결이 여러 개인 경우 매우 편리하게 문서에 표시합니다. 그 후, 각 연결에 대해 논리적 if 구성을 사용하여 고유한 필터를 작성할 수 있습니다.

파일 :

input {
  file {
    path => "/var/log/openvas_report/*"
    type => "openvas"
    start_position => "beginning"
    }
}

설정 설명:
경로 => "/var/log/openvas_report/*"
파일을 읽어야 하는 디렉터리를 나타냅니다.

유형 => "오픈바스"
이벤트 유형.

start_position => "시작"
파일을 변경할 때 전체 파일을 읽으며, “end”를 설정하면 시스템은 파일 끝에 새 레코드가 나타날 때까지 기다립니다.

임원:

input {
  exec {
    command => "ls -alh"
    interval => 30
  }
}

이 입력을 사용하면 (유일한!) 셸 명령이 실행되고 해당 출력이 로그 메시지로 변환됩니다.

명령 => "ls -alh"
우리가 관심 있는 출력의 명령입니다.

간격 => 30
명령 호출 간격(초)입니다.

방화벽으로부터 로그를 받기 위해 필터를 등록합니다. TCP 또는 udp, 로그가 Logstash로 전송되는 방식에 따라 다릅니다.

로그 메시지의 모양을 이해하기 위해 디버그 모드에서 Logstash 구성 파일의 출력을 구성합니다.

INPUT을 구성한 후에는 로그 메시지의 모양과 로그 필터(파서)를 구성하는 데 사용해야 하는 방법을 이해해야 합니다.

이를 위해 원본 메시지를 보기 위해 결과를 stdout으로 출력하는 필터를 사용합니다. 현재 전체 구성 파일은 다음과 같습니다.

input 
{
         tcp 
         {
                port => 5555
  	  	type => "checkpoint"
    		mode => "server"
                host => “10.10.1.205”
   	 }
}

output 
{
	if [type] == "checkpoint" 
       {
		stdout { codec=> json }
	}
}

다음 명령을 실행하여 확인합니다.
sudo /usr/share/logstash/bin//logstash -f /etc/logstash/conf.d/checkpoint.conf
결과를 보면 그림을 클릭할 수 있습니다.

복사하면 다음과 같이 됩니다.

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,

이러한 메시지를 보면 로그가 필드 = 값 또는 키 = 값과 유사하다는 것을 알 수 있습니다. 이는 kv라는 필터가 적합하다는 의미입니다. 각 특정 사례에 적합한 필터를 선택하려면 기술 문서에서 해당 필터를 숙지하거나 친구에게 문의하는 것이 좋습니다.

필터 설정

kv를 선택한 마지막 단계에서 이 필터의 구성은 아래와 같습니다.

filter {
if [type] == "checkpoint"{
	kv {
		value_split => "="
		allow_duplicate_values => false
	}
}
}

필드와 값을 나눌 기호("=")를 선택합니다. 로그에 동일한 항목이 있으면 데이터베이스에 하나의 인스턴스만 저장합니다. 그렇지 않으면 동일한 값의 배열로 끝납니다. 즉, "foo = some foo=some"이라는 메시지가 있으면 foo만 씁니다. = 일부.

ElasticSearch에서 올바른 출력 설정

필터가 구성된 후 데이터베이스에 로그를 업로드할 수 있습니다. 탄성 검색:

output 
{
if [type] == "checkpoint"
{
 	elasticsearch 
        {
		hosts => ["10.10.1.200:9200"]
		index => "checkpoint-%{+YYYY.MM.dd}"
    		user => "tssolution"
    		password => "cool"
  	}
}
}

문서가 체크포인트 유형으로 서명된 경우 기본적으로 포트 10.10.1.200에서 9200의 연결을 허용하는 Elasticsearch 데이터베이스에 이벤트를 저장합니다. 각 문서는 특정 인덱스에 저장됩니다. 이 경우 "체크포인트-" + 현재 시간 날짜 인덱스에 저장됩니다. 각 인덱스는 특정 필드 집합을 가질 수 있거나 메시지에 새 필드가 나타날 때 자동으로 생성되며, 필드 설정 및 해당 유형은 매핑에서 볼 수 있습니다.

인증을 구성한 경우(나중에 살펴보겠습니다) 특정 인덱스에 쓰기 위한 자격 증명을 지정해야 합니다. 이 예에서는 비밀번호가 "cool"인 "tssolution"입니다. 특정 인덱스에만 로그를 기록하고 더 이상 로그를 기록하지 않도록 사용자 권한을 차별화할 수 있습니다.

로그스태시를 실행하세요.

Logstash 구성 파일:

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"
  	}
}
}

구성 파일의 정확성을 확인합니다.
/usr/share/logstash/bin//logstash -f checkpoint.conf


Logstash 프로세스를 시작합니다.
sudo systemctl start logstash

프로세스가 시작되었는지 확인합니다.
sudo systemctl 상태 로그스태시

소켓이 작동하는지 확인해 보겠습니다.
netstat -nat |grep 5555

Kibana에서 로그를 확인합니다.

모든 것이 실행된 후 Kibana - Discover로 이동하여 모든 것이 올바르게 구성되었는지 확인하세요. 그림을 클릭할 수 있습니다!

모든 로그가 제자리에 있으며 모든 필드와 해당 값을 볼 수 있습니다!

결론

Logstash 구성 파일을 작성하는 방법을 살펴본 결과 모든 필드와 값에 대한 파서를 얻었습니다. 이제 특정 필드를 검색하고 플로팅하는 작업을 수행할 수 있습니다. 다음 과정에서는 Kibana의 시각화를 살펴보고 간단한 대시보드를 만들어 보겠습니다. 특정 상황(예: 필드 값을 숫자에서 단어로 바꾸려는 경우)에서는 Logstash 구성 파일을 지속적으로 업데이트해야 한다는 점을 언급할 가치가 있습니다. 후속 기사에서 우리는 이 작업을 지속적으로 수행할 것입니다.

그러니 계속 지켜봐 주시기 바랍니다(텔레그램, 페이스북, VK, TS 솔루션 블로그), Yandex.Dzena.

출처 : habr.com