ELK收集Oracle审计日志

迈不过友情╰ 2023-01-17 03:22 271阅读 0赞

一、开启审计，记录sql

  1. --Oracle审计
  2. https://max.book118.com/html/2016/1214/72170219.shtm
  4. --开启审计
  5. alter system set audit_sys_operations=TRUE scope=spfile;
  6. alter system set audit_trail=db,extended scope=spfile;
  7. --重启实例
  8. shutdown immediate
  9. startup
  10. --查看审计状态
  11. SYS@orcl>show parameter audit;
  13. NAME                     TYPE     VALUE
  14. ------------------------------------ ----------- ------------------------------
  15. audit_file_dest              string     /u01/app/oracle/admin/orcl/adump
  16. audit_sys_operations         boolean     TRUE
  17. audit_syslog_level             string
  18. audit_trail                     string     DB, EXTENDED

二、迁移审计表空间

  1. --迁移审计表到ogg的表空间
  2. BEGIN
  3. DBMS_AUDIT_MGMT.set_audit_trail_location(
  4. audit_trail_type => DBMS_AUDIT_MGMT.AUDIT_TRAIL_AUD_STD,
  5. audit_trail_location_value => 'OGG');
  6. END;
  7. /
  9. BEGIN
  10. DBMS_AUDIT_MGMT.set_audit_trail_location(
  11. audit_trail_type => DBMS_AUDIT_MGMT.AUDIT_TRAIL_FGA_STD,
  12. audit_trail_location_value => 'OGG');
  13. END;
  14. /
  16. --验证表空间
  17. SELECT table_name, tablespace_name FROM dba_tables WHERE table_name IN ('AUD$', 'FGA_LOG$');
  18. select TABLE_NAME, SEGMENT_NAME, TABLESPACE_NAME from dba_lobs where table_name in ('AUD$', 'FGA_LOG$');
  20. --审计段大小统计
  21. SELECT owner
  22.       ,table_name
  23.       ,SUM(decode(seg_type, 'table', size_mb)) tab_size_mb
  24.       ,SUM(decode(seg_type, 'index', size_mb)) idx_size_mb
  25.       ,SUM(decode(seg_type, 'lob', size_mb)) lob_size_mb
  26.       ,SUM(size_mb)
  27.   FROM (SELECT /*+ rule */  t.owner
  28.               ,t.table_name
  29.               ,SUM(s.bytes) / 1024 / 1024 size_mb
  30.               ,'table' seg_type
  31.           FROM dba_segments s, dba_tables t
  32.          WHERE s.owner = t.owner
  33.            AND s.segment_name = t.table_name
  34.          GROUP BY t.owner, t.table_name
  35.         UNION ALL
  36.         SELECT /*+ rule */ l.owner
  37.               ,l.table_name
  38.               ,SUM(s.bytes) / 1024 / 1024 size_mb
  39.               ,'lob' seg_type
  40.           FROM dba_segments s, dba_lobs l
  41.          WHERE s.owner = l.owner
  42.            AND s.segment_name = l.segment_name
  43.          GROUP BY l.owner, l.table_name
  44.         UNION ALL
  45.         SELECT /*+ rule */ i.table_owner owner
  46.               ,i.table_name
  47.               ,SUM(s.bytes) / 1024 / 1024 size_mb
  48.               ,'index' seg_type
  49.           FROM dba_segments s, dba_indexes i
  50.          WHERE s.owner = i.owner
  51.            AND s.segment_name = i.index_name
  52.          GROUP BY i.table_owner, i.table_name)
  53.  WHERE owner = 'SYS'
  54.    AND table_name = 'AUD$'
  55.    group by owner, table_name;

Oracle审计相关内容参考:

https://blog.csdn.net/demonson/article/details/116146370

三、根据需求设计审计规则

  1. 需求:
  2.   要求ELK收集现有数据库所有sql语句，但由于DML语句过于频繁，只收集数据库所有DDL语句（不管语句是否执行成功，都收集上）；

根据以上需求和上面oracle审计技术文档，判断为语句级审计满足需求： 需要审计项目：

" class="reference-link">watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RlbW9uc29u_size_16_color_FFFFFF_t_70

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RlbW9uc29u_size_16_color_FFFFFF_t_70 1

  1. 先清空oracle默认开启的审计项
  3. select * from sys.audit$;
  6. --sys用户登录
  7. truncate table audit$;

在执行下面语句级审计DDL

  1. 审计语句:
  3. AUDIT DATABASE LINK;
  4. AUDIT INDEX;
  5. AUDIT PROCEDURE;
  6. AUDIT SEQUENCE;
  7. AUDIT SYSTEM GRANT;
  8. AUDIT TABLE;
  9. AUDIT TABLESPACE;
  10. AUDIT TRIGGER;
  11. AUDIT USER; 
  12. AUDIT TYPE;
  13. AUDIT VIEW;
  14. AUDIT ALTER SEQUENCE;
  15. AUDIT ALTER TABLE;

重启数据库

  1. shut immediate
  2. startup

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RlbW9uc29u_size_16_color_FFFFFF_t_70 2

  1. 涉及到的SQL:
  2. select * from sys.audit$;
  3. select * from sys.aud$;
  4. select * from dba_priv_audit_opts;
  5. select * from dba_stmt_audit_opts;
  6. select * from sys.AUDIT$ a inner join sys.STMT_AUDIT_OPTION_MAP b on a.option#=b.OPTION#;
  8. select userid,userhost, to_char(substr(SQLTEXT,1,3000)) as sql,action#,returncode,to_char(NTIMESTAMP#,'yyyy-mm-dd hh24:mi:ss'),
  9. SQLBIND,SQLTEXT from sys.aud$  t where userid='GGS'order by t.ntimestamp# desc;
  11. 取消审计语句:
  13. NOAUDIT DATABASE LINK;
  14. NOAUDIT INDEX;
  15. NOAUDIT PROCEDURE;
  16. NOAUDIT SEQUENCE;
  17. NOAUDIT SYSTEM GRANT;
  18. NOAUDIT TABLE;
  19. NOAUDIT TABLESPACE;
  20. NOAUDIT TRIGGER;
  21. NOAUDIT USER;
  22. NOAUDIT TYPE;
  23. NOAUDIT VIEW;
  24. NOAUDIT ALTER SEQUENCE;
  25. NOAUDIT ALTER TABLE;
  27. 取消所有审计语句:
  29. NOAUDIT ALL;

#取消审计等操作若是不生效，测试重新配置审计打开并重启数据库能有效解决

四、ELK抓取审计语句

1.sys用户下创建视图

  1. create or replace view audit_sql_vw as
  2. select to_char(t.NTIMESTAMP# + 1 / 3, 'yyyy-mm-dd hh24:mi:ss') auditime, --审计时间
  3.        t.userid as username, --用户名称
  4.        t.userhost as hostname, --客户端主机名
  5.        b.name as sqltype, --语句类别
  6.        t.obj$name as objname, --对象名称
  7.        t.returncode, --返回值，0表示执行成功
  8.        to_char(substr(t.SQLTEXT, 1, 3000)) as sql --审计sql
  9.   from sys.aud$ t
  10.   left join sys.AUDIT_ACTIONS b
  11.     on t.action# = b.action;

2.编写python，将数据库中审计日志读取到磁盘上

  1. #创建日志目录
  3. su - oracle
  4. mkdir /u01/app/oracle/admin/orcl/audit
  6. vim check_audit.py
  8. #!/usr/bin/python
  9. #coding=utf-8
  10. import os
  11. import cx_Oracle
  13. os.environ['ORACLE_HOME'] = '/u01/app/oracle/product/11.2.0/db_1'
  14. os.environ['ORACLE_SID'] = 'orcl'
  15. os.environ['PATH']
  17. #连接数据库，查视图
  18. def check_audit_qurey():
  19.     conn = cx_Oracle.connect('sys','*******','192.168.1.96:14320/orcl',mode=cx_Oracle.SYSDBA)
  20.     cursor=conn.cursor()
  21.     lists = []
  22.     try:
  23.         cursor.execute ("select * from audit_sql_vw")  
  24.         #print("连接成功！")
  25.         lists = cursor.fetchall()  
  26.     except Exception:
  27.         print("connenct oracle error,dblink error!",Exception)
  28.     finally:
  29.         #每次运行脚本后，清空数据库中审计日志，防止占用空间太大
  30.         cursor.execute ("truncate table sys.aud$")
  31.         cursor.close()  
  32.         conn.close()  
  33.     msg = ''
  34.     if len(lists):
  35.         for i in lists:
  36.             msg = msg +  ",".join(map(str, i)) + '\n'
  37.         #print(msg)
  38.         #将文件输出到文件
  39.         f=open('/u01/app/oracle/admin/orcl/audit/audit.log','w+')
  40.         f.write(msg)
  41.         f.close()
  44. def main():
  45.     check_audit_qurey()
  46. if __name__ == "__main__":
  47.     main()

3.每半小时执行一次脚本

  1. crontab -e
  3. */30 * * * * /usr/bin/python /u01/app/oracle/admin/orcl/audit/check_audit.py > /u01/app/oracle/admin/orcl/audit/exec_check_audit.log 2>&1

4.安装filebeat读取oracle审计日志

  1. 网上下载filebeat的rpm包
  3. rpm -ivh filebeat-7.12.0-x86_64.rpm 
  4. warning: filebeat-7.12.0-x86_64.rpm: Header V4 RSA/SHA512 Signature, key ID d88e42b4: NOKEY
  5. Preparing...                ########################################### [100%]
  6.    1:filebeat               ########################################### [100%]
  8. #修改配置文件
  9. vim /etc/filebeat/filebeat.yml
  11. filebeat.inputs:
  13. #oracle_audit.log 收集审计SQL(每半小时会自动触发一次)
  14. - type: log
  15.   enabled: true
  16.   paths:
  17.     - /u01/app/oracle/admin/orcl/audit/audit.log
  18.   tags: ["oracle-audit-log"]
  19.   multiline.pattern:  '\d{4}-\d{2}-\d{2}'
  20.   multiline.negate: true
  21.   multiline.match: after
  22.   fields:
  23.       log_source: oracle-audit-log
  25. filebeat.config.modules:
  26.   path: ${path.config}/modules.d/*.yml
  27.   reload.enabled: false
  28. setup.template.settings:
  29.   index.number_of_shards: 3
  30. setup.kibana:
  32. output.logstash:
  33.   hosts: ["192.168.1.207:55051"]
  35. #启动filebeat
  36. service filebeat start
  37. service filebeat status
  38. service filebeat stop
  40. #日志查看
  41. tail -f /var/log/filebeat/filebeat

5.创建logstash配置文件，重启远程logstash

  1. #配置logstash内置正则字段
  2. vim /usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-patterns-core-4.1.2/patterns/grok-patterns
  4. 添加
  6. ORACLE_TIME (Mon|Tue|Wed|Thu|Fri|Sat|Sun)\s(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)\s(\d{2})\s(\d{2}):(\d{2}):(\d{2})\s(\d{4})
  7. DB_LOGCONTENT .*
  8. SQL_ID [A-Za-z0-9]+
  9. ORA_MACHINE .*?
  10. ORA_USER .*?
  11. ORA_TYPE .*?
  12. ORA_OBJECT .*?
  13. ORA_SQL .*
  15. logstash配置文件创建
  16. vim /etc/logstash/conf.d/oracle-log.conf
  18. vim /etc/logstash/conf.d/oracle-log.conf
  20. input {
  21.   beats {
  22.     port => 55051
  23.   }
  24. }
  26. filter{
  27.   if "oracle-log" in [tags] {
  28.     mutate { add_field => {  "clienthost" => "%{[beat][hostname]}" } }
  29.     grok {
  30.         match => {
  31.             "message" =>  "%{ORACLE_TIME:datetime}\n%{DB_LOGCONTENT:logmessage}"
  32.         }
  33.     }
  34.   }
  36.  if "oracle-slow-log" in [tags] {
  37.  grok {
  38.         match => [ "message" , "%{TIMESTAMP_ISO8601:time},%{SQL_ID:sql_id},%{NUMBER:elapsed_time:float},%{NUMBER:cpu_time:float},%{NUMBER:iowait_time:float},%{NUMBER:gets:int},%{NUMBER:reads:int},%{NUMBER:rows:int},%{NUMBER:cluster_wait_time:float},%{NUMBER:execs:int},%{NUMBER:elpe_time:float},%{ORA_MACHINE:machine},%{ORA_USER:username},%{ORA_SQL:sql}" ]
  40.     }
  41. date{
  42.         match=>["time","YYYY-MM-dd HH:mm:ss"]
  43.         target=>"snapshot_time"
  44.     }
  45.  }
  48.   if "oracle-audit-log" in [tags] {
  49.  grok {
  50.         match => [ "message" , "%{TIMESTAMP_ISO8601:time},%{ORA_USER:user_name},%{ORA_MACHINE:machine},%{ORA_TYPE:sql_type},%{ORA_OBJECT:object_name},%{NUMBER:return_code:int},%{ORA_SQL:sql}" ]
  52.     }
  53.  date{
  54.         match=>["time","YYYY-MM-dd HH:mm:ss"]
  55.         target=>"audit_time"
  56.     }
  57.  }
  59. }
  62. output {
  63.   if "oracle-log" in [tags] {
  64.     elasticsearch {
  65.       hosts => ["192.168.1.207:9200"]
  66.       manage_template => false
  67.       index => "oracle-log-%{+YYYY.MM}"
  68.    }}
  69.   if "oracle-slow-log" in [tags] {
  70.     elasticsearch {
  71.       hosts => ["192.168.1.207:9200"]
  72.       manage_template => false
  73.       index => "oracle-slow-log-%{+YYYY.MM}"
  74.    }
  75.  }
  76.  if "oracle-audit-log" in [tags] {
  77.     elasticsearch {
  78.       hosts => ["192.168.1.207:9200"]
  79.       manage_template => false
  80.       index => "oracle-audit-log-%{+YYYY.MM}"
  81.    }
  82.  }
  83. }
  85. #重启logstash
  86. systemctl restart logstash

6.kabana创建索引，查看审计状态

kabana创建索引

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RlbW9uc29u_size_16_color_FFFFFF_t_70 3

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RlbW9uc29u_size_16_color_FFFFFF_t_70 4

发表评论

表情：
评论列表 （有 0 条评论，271人围观）

还没有评论，来说两句吧...

相关阅读