4.ELK

[root@logstash etc]# cat /etc/rsyslog.conf
# rsyslog configuration file
 
# For more information see /usr/share/doc/rsyslog-*/rsyslog_conf.html
# or latest version online at http://www.rsyslog.com/doc/rsyslog_conf.html 
# If you experience problems, see http://www.rsyslog.com/doc/troubleshoot.html
 
#### MODULES ####
 
module(load="imuxsock"    # provides support for local system logging (e.g. via logger command)
       SysSock.Use="off") # Turn off message reception via local log socket; 
                          # local messages are retrieved through imjournal now.
module(load="imjournal"             # provides access to the systemd journal
       UsePid="system" # PID nummber is retrieved as the ID of the process the journal entry originates from
       StateFile="imjournal.state") # File to store the position in the journal
#module(load="imklog") # reads kernel messages (the same are read from journald)
#module(load="immark") # provides --MARK-- message capability
 
# Provides UDP syslog reception
# for parameters see http://www.rsyslog.com/doc/imudp.html
module(load="imudp") # needs to be done just once
input(type="imudp" port="514")
 
# Provides TCP syslog reception
# for parameters see http://www.rsyslog.com/doc/imtcp.html
module(load="imtcp") # needs to be done just once
input(type="imtcp" port="514")
 
#### GLOBAL DIRECTIVES ####
 
# Where to place auxiliary files
global(workDirectory="/var/lib/rsyslog")
 
# Use default timestamp format
module(load="builtin:omfile" Template="RSYSLOG_TraditionalFileFormat")
 
$template NetworkTemplate,"%timereported:::date-rfc3339% ip-address:%fromhost-ip% log-severity:%syslogseverity% log-facility:%syslogfacility% msg:%msg:::sp-if-no-1st-sp%%msg:::drop-last-lf%\n"
 
$template H3cDir, "/var/log/network/%fromhost-ip%/%fromhost-ip%.log"
 
$template HCI_Log, "/var/log/hci/hci-%$YEAR%-%$MONTH%-%$DAY%.log"
 
 
# Include all config files in /etc/rsyslog.d/
include(file="/etc/rsyslog.d/*.conf" mode="optional")
 
#### RULES ####
 
# Log all kernel messages to the console.
# Logging much else clutters up the screen.
#kern.*                                                 /dev/console
if ($fromhost-ip == '10.0.13.114' or $fromhost-ip == '10.0.13.113') then {
    action(type="omfile" file="/var/log/ips/ips.log" template="NetworkTemplate")
    & stop
}
 
if ($fromhost-ip == '10.0.12.31' or $fromhost-ip == '10.0.12.32' or $fromhost-ip == '10.0.12.34' or $fromhost-ip == '10.0.12.33') then {
    action(type="omfile" file="/var/log/hci-test/HCI_10.0.12.30.log" template="NetworkTemplate")
    & stop
}
 
if ($fromhost-ip == '10.0.12.11' or $fromhost-ip == '10.0.12.12' or $fromhost-ip == '10.0.12.15' or $fromhost-ip == '10.0.12.14' or $fromhost-ip == '10.0.12.16' or $fromhost-ip == '10.0.12.17') then {
    action(type="omfile" file="/var/log/hci/HCI_10.0.12.10.log" template="NetworkTemplate")
    & stop
}
 
if ($fromhost-ip == '10.0.13.110' or $fromhost-ip == '10.0.13.111') then {
    action(type="omfile" file="/var/log/waf/waf.log" template="NetworkTemplate")
    & stop
} 
 
if ($fromhost-ip == '10.0.10.2') then {
   action(type="omfile" file="/var/log/ac/ac_10.0.10.2.log" template="NetworkTemplate")
}
 
if ($fromhost-ip == '10.0.20.2') then {
   action(type="omfile" file="/var/log/ad/ad_10.0.20.2.log" template="NetworkTemplate")
}
 
if ($fromhost-ip == '10.10.8.30') then {
   action(type="omfile" file="/var/log/edr/edr_10.10.8.30.log" template="NetworkTemplate")
}
 
if ($fromhost-ip == '10.11.8.11') then {
   action(type="omfile" file="/var/log/mbs/mbs_10.11.8.11.log" template="NetworkTemplate")
}
 
 
local5.* action(type="omfile" FileCreateMode="0755" DirCreateMode="0755" dynaFile="H3cDir" template="NetworkTemplate")
 
 
 
# Log anything (except mail) of level info or higher.
# Don't log private authentication messages!
#*.info;mail.none;authpriv.none;cron.none;               /var/log/messages
mail.none;authpriv.none;cron.none;               /var/log/messages
 
# The authpriv file has restricted access.
authpriv.*                                              /var/log/secure
 
# Log all the mail messages in one place.
mail.*                                                  -/var/log/maillog
 
 
# Log cron stuff
cron.*                                                  /var/log/cron
 
# Everybody gets emergency messages
*.emerg                                                 :omusrmsg:*
 
# Save news errors of level crit and higher in a special file.
uucp,news.crit                                          /var/log/spooler
 
# Save boot messages also to boot.log
local7.*                                                /var/log/boot.log
 
 
# ### sample forwarding rule ###
#action(type="omfwd"  
# An on-disk queue is created for this action. If the remote host is
# down, messages are spooled to disk and sent when it is up again.
#queue.filename="fwdRule1"       # unique name prefix for spool files
#queue.maxdiskspace="1g"         # 1gb space limit (use as much as possible)
#queue.saveonshutdown="on"       # save messages to disk on shutdown
#queue.type="LinkedList"         # run asynchronously
#action.resumeRetryCount="-1"    # infinite retries if host is down
# Remote Logging (we use TCP for reliable delivery)
# remote_host is: name/ip, e.g. 192.168.0.1, port optional e.g. 10514
#Target="remote_host" Port="XXX" Protocol="tcp")

[root@logstash ~]# cd /etc/logrotate.d/
[root@logstash logrotate.d]# ls
ac_log  ad_log  btmp  chrony  dnf  edr_log  firewalld  hci_log  hci_log_test  kvm_stat  mbs_log  network_log  psacct  samba  sssd  syslog  waf  wtmp
[root@logstash logrotate.d]# 
[root@logstash logrotate.d]# cat edr_log 
/var/log/edr/edr_10.10.8.30.log {               # 作用域, 即要操作的日志文件绝对路径
    daily                  # 按天切割日志，也可以是 weekly, monthly 等
    rotate 7
    compress               # 使用 gzip 压缩旧日志
    delaycompress          # 延迟压缩，先不压缩最新一份日志，直到下一次切割
    missingok              # 如果日志文件丢失，不生成错误信息
    notifempty             # 如果日志文件为空，则不进行切割
    copytruncate
    create 600 root root
}

[root@logstash logrotate.d]# cat /etc/filebeat/filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/ips/ips.log
  fields:
    source: ips
  tags: ["ips"]
 
- type: log
  enabled: true
  paths:
    - /var/log/hci/HCI_10.0.12.10.log
  fields:
    source: hci
  tags: ["hci"]
 
- type: log
  enabled: true
  paths:
    - /var/log/waf/waf.log
  fields:
    source: waf
  tags: ["waf"]  
 
- type: log
  enabled: true
  paths:
    - /var/log/hci-test/HCI_10.0.12.30.log
  fields:
    source: hci
  tags: ["hci_test"]
 
- type: log
  enabled: true
  paths:
    - /var/log/network/*/*.log
  tags: ["h3c"]
 
- type: log
  enabled: true
  paths:
    - /var/log/ad/ad_10.0.20.2.log
  tags: ["ad"]
 
- type: log
  enabled: true
  paths:
    - /var/log/ac/ac_10.0.10.2.log
  tags: ["ac"]
 
- type: log
  enabled: true
  paths:
    - /var/log/edr/edr_10.10.8.30.log
  tags: ["edr"]
  fields:
    source: edr
- type: log
  enabled: true
  paths:
    - /var/log/mbs/mbs_10.11.8.11.log
  tags: ["mbs"]
  fields:
    source: mbs
 
# 设置模板
setup.template.name: "filebeat"
setup.template.pattern: "*" 
setup.template.enabled: true
 
# 模板设置
setup.template.settings:
  index.number_of_shards: 1
  index.number_of_replicas: 0  # 可选：设置副本数量
 
output.elasticsearch:
  # Elasticsearch 的主机地址
  hosts: ["http://10.11.8.80:9200"]
 
  # 设置用户名和密码
  username: "elastic"
  password: "66Mqgfj6PC=CfN4FCA6y"
  index: "network-%{+yyyy.MM.dd}"
 
 
 
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
 
# 处理器配置
processors:
  - drop_fields:
      fields: 
        - agent.ephemeral_id
        - agent.hostname
        - agent.id
        - agent.type
        - agent.version
        - ecs.version
        - host.name
        - input.type
        - log.offset
        - log.file.path
        - agent.hostname
        - agent.name

 
[root@localhost ~]# cat /etc/filebeat/filebeat.yml 
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /home/amc/aes/build/logs/manatee-system/error.log
  multiline.pattern: '^\d{4}-\d{2}-\d{2}'
  multiline.negate: true    
  multiline.match: after
  multiline.max_lines: 100
  tags: ["aes", "pre", "10.10.10.101", "error"]
 
- type: log
  enabled: true
  paths:
    - /home/amc/aes/build/logs/manatee-system/info.log
  multiline.pattern: '^\d{4}-\d{2}-\d{2}'
  multiline.negate: true    
  multiline.match: after
  multiline.max_lines: 100
  tags: ["aes", "pre", "10.10.10.101", "info"]
 
 
# 设置模板
setup.template.name: "filebeat"
setup.template.pattern: "*" # 使用数组来匹配多个模式
setup.template.enabled: true
 
# 模板设置
setup.template.settings:
  index.number_of_shards: 1
  index.number_of_replicas: 0  # 可选：设置副本数量
 
# 输出到 Elasticsearch
output.elasticsearch:
  # Elasticsearch 的主机地址
  hosts: ["http://10.11.8.80:9200"]
 
  # 设置用户名和密码
  username: "elastic"
  password: "66Mqgfj6PC=CfN4FCA6y"
  index: "filebeat-%{+yyyy.MM.dd}"
 
 
 
# 处理器配置
processors:
  - drop_fields:
      fields: 
        - agent.ephemeral_id
        - agent.hostname
        - agent.id
        - agent.type
        - agent.version
        - ecs.version
        - host.name
        - input.type
        - log.offset
        - log.file.path
        - agent.hostname
        - agent.name
        - message.keyword
        - log.flags

[root@localhost elasticsearch]# cat docker-compose.yml 
services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.14.3
    container_name: elasticsearch
    hostname: elasticsearch
    environment:
      - "ES_JAVA_OPTS=-Xms2g -Xmx2g"
      - "discovery.type=single-node"
      - "LANG=C.UTF-8"
      - "LC_ALL=C.UTF-8"
      - "TZ=Asia/Shanghai"
    restart: always
    volumes:
      - './config:/usr/share/elasticsearch/config'
      - './data:/usr/share/elasticsearch/data'
      - './plugins:/usr/share/elasticsearch/plugins'
      - './logs:/usr/share/elasticsearch/logs'
    ports:
      - "9300:9300"
      - "9200:9200"
    user: "1003:1003"
    networks:
      - network
networks:
  network:
    external: true

[root@localhost kibana]# cat docker-compose.yml 
services:
  kibana:
    image: docker.elastic.co/kibana/kibana:8.14.3
    container_name: kibana
    hostname: kibana
    restart: always
    environment:
     - "I18N_LOCALE=zh-CN"
     - "TZ=Asia/Shanghai"
    volumes:
      - './config:/usr/share/kibana/config'
      - './data:/usr/share/kibana/data'
      - './logs:/usr/share/kibana/logs'
    ports:
      - "5601:5601"
    user: "1001:1001"
    networks:
      - network
networks:
  network:
    external: true

[root@localhost elastalert2]# cat docker-compose.yml 
version: "2.21.0"
services:
  elastalert2:
    image: ghcr.io/jertel/elastalert2/elastalert2:latest
    container_name: elastalert2
    hostname: elastalert2
    restart: always
    volumes:
      - './conf/config.yaml:/opt/elastalert/config.yaml'
      - './rules:/opt/elastalert/rules'
      - './cert:/etc/elastalert/cert'
    networks:
      - elastic
networks:
  elastic:
    external: true

[root@localhost rules]# cat aes_error_elart.yaml 
name: "dingtalk-aes" #规则的名称。
type: "any" # 规则类型为频率，表示要检测的事件是在一定时间范围内出现的次数。
index: "filebeat*" # 查询日志所在的索引
is_enabled: true
timestamp_field: "@timestamp"
timestamp_type: "iso"
use_strftime_index: false
alert_text_type: alert_text_only
# 下面是告警模板
alert_text: |
  生态圈日志告警
  服务器: {0}
  触发次数: {1}
  匹配次数: {2}
  日志信息: {3}
alert_text_args: # 告警模板中用到的参数
  - tags
  - num_hits
  - num_matches
  - message
filter:
- term:
    tags: "prod"
- term:
    tags: "error"
- term:
    tags: "aes"
alert:
  - "dingtalk" # 告警类型
dingtalk_access_token: "0d6e66f4dcc7d9febca04134858b102d2fb9c87853f85707be4e44bba14e6e38"
dingtalk_msgtype: "text" # 消息类型

[root@localhost network]# cat ips_alter.yaml 
name: "dingtalk-ips-alert"
type: "frequency" # 规则类型为频率，表示要检测的事件是在一定时间范围内出现的次数。
index: "network*" # 查询日志所在的索引
is_enabled: true
num_events: 1 # 出现几次就告警
timeframe:
  minutes: 1 # 1分钟 出现了 num_events次 匹配记录，就告警
realert:
  minutes: 1 # 1分钟内忽略重复告警
timestamp_field: "@timestamp"
timestamp_type: "iso"
use_strftime_index: false
alert_text_type: alert_text_only
 
alert_text: |
  ips告警
  服务器: {0}
  触发次数: {1}
  匹配次数: {2}
  日志信息: {3}
alert_text_args: # 告警模板中用到的参数
  - tags
  - num_hits
  - num_matches
  - message
 
filter:
- term:
    tags: "ips"
 
alert:
  - "dingtalk"
  
dingtalk_access_token: "7d902107140f9dad650dd96d55e81dd918696ca2e31612a4814066cebd1387c3"
dingtalk_msgtype: "text" # 消息类型

[root@localhost network]# cat mbs_alter.yaml 
name: "dingtalk-mbs-alert"
type: "frequency" # 规则类型为频率，表示要检测的事件是在一定时间范围内出现的次数。
index: "network*" # 查询日志所在的索引
is_enabled: true
num_events: 1 # 出现几次就告警
timeframe:
  minutes: 1 # 1分钟 出现了 num_events次 匹配记录，就告警
realert:
  minutes: 1 # 1分钟内忽略重复告警
timestamp_field: "@timestamp"
timestamp_type: "iso"
use_strftime_index: false
alert_text_type: alert_text_only
 
alert_text: |
  mbs告警
  服务器: {0}
  触发次数: {1}
  匹配次数: {2}
  日志信息: {3}
alert_text_args: # 告警模板中用到的参数
  - tags
  - num_hits
  - num_matches
  - message
 
filter:
 - bool:
      must:
        - term:
            tags: "mbs"
        - query_string:
            query: "message:*attackIp*"
 
alert:
  - "dingtalk"
  
    #dingtalk_access_token: "567ca64e2b84eab0847ffc20a794ccc558db0f0123b7bfeb883ca57a1628b947"
dingtalk_access_token: "75dc8bd3727cf7dc28850c2f8de0d3eacfacad4f20ec2a33bb16c1e97aaa3f59"
dingtalk_msgtype: "text" # 消息类型

EFK

1.rsyslog 收集设备日志

2.filebeat推送到Elasticsearch

3.filebate收集应用日志推送到Elasticsearch

4.部署elasticsearch

5.部署kibana

6.部署elastalert2