Ubuntu Server系统深度审计指南：从基础到生产环境实践

struct audit_krule {
    u32			pflags;
    u32			action;
    u32			mask[AUDIT_BITMASK_SIZE];
    u32			buflen;
    u32			field_count;
    struct audit_field *fields;
    struct audit_field *inode_f;
    struct audit_watch *watch;
};

# 更新系统并安装审计组件
sudo apt update && sudo apt upgrade -y
sudo apt install auditd audispd-plugins audit-viewer -y

# 验证安装完整性
sudo systemctl status auditd
sudo auditctl -s
sudo ausearch --version

# 安装增强插件
sudo apt install audispd-plugins-extra \
    auditd-exporter \
    python3-audit -y

# /etc/audit/auditd.conf 生产环境配置
log_file = /var/log/audit/audit.log
log_format = ENRICHED
log_group = adm
priority_boost = 4
flush = INCREMENTAL_ASYNC
freq = 50
max_log_file = 100
num_logs = 10
max_log_file_action = ROTATE
space_left = 75
space_left_action = SYSLOG
action_mail_acct = root
admin_space_left = 50
admin_space_left_action = SUSPEND
disk_full_action = SUSPEND
disk_error_action = SUSPEND
use_libwrap = yes
tcp_listen_port = 60
tcp_max_per_addr = 1
transport = TCP
q_depth = 400
overflow_action = SYSLOG
max_restarts = 10
plugin_dir = /etc/audit/plugins.d

# /etc/audit/rules.d/audit.rules
## 首先清空现有规则
-D

## 不可修改的审计规则
-e 2
-f 2
--backlog_wait_time 60000
-b 8192

## 关键文件系统监控
-w /etc/passwd -p wa -k identity
-w /etc/group -p wa -k identity
-w /etc/shadow -p wa -k identity
-w /etc/gshadow -p wa -k identity
-w /etc/sudoers -p wa -k privilege
-w /etc/ssh/sshd_config -p wa -k sshd
-w /etc/audit/ -p wa -k audit
-w /etc/default/grub -p wa -k boot
-w /boot/ -p wa -k boot

## 系统调用审计
-a always,exit -F arch=b64 -S execve -k process
-a always,exit -F arch=b64 -S bind -k network
-a always,exit -F arch=b64 -S connect -k network
-a always,exit -F arch=b64 -S accept -k network
-a always,exit -F arch=b64 -S socket -k network

## 特权操作监控
-a always,exit -F arch=b64 -S chmod -F auid>=1000 -k perm_mod
-a always,exit -F arch=b64 -S chown -F auid>=1000 -k perm_mod
-a always,exit -F arch=b64 -S setxattr -F auid>=1000 -k perm_mod
-a always,exit -F arch=b64 -S mount -F auid>=1000 -k mount
-a always,exit -F arch=b64 -S ptrace -k tracing

## 用户会话审计
-a always,exit -F arch=b64 -S execve -F uid=0 -k admin_actions
-a always,exit -F arch=b64 -S open -F success=0 -k access_denied
-a always,exit -F arch=b64 -S creat -F success=0 -k access_denied
-a always,exit -F arch=b64 -S openat -F success=0 -k access_denied
-a always,exit -F arch=b64 -S truncate -F success=0 -k access_denied

# 关键系统文件监控
## 系统二进制文件
-w /bin -p wa -k system_bin
-w /sbin -p wa -k system_bin
-w /usr/bin -p wa -k system_bin
-w /usr/sbin -p wa -k system_bin
-w /usr/local/bin -p wa -k local_bin

## 配置文件目录
-w /etc/ -p wa -k etc_config
-w /etc/network/ -p wa -k network_config
-w /etc/ssl/ -p wa -k ssl_config
-w /etc/apache2/ -p wa -k web_config
-w /etc/mysql/ -p wa -k db_config

## 临时目录监控（异常活动检测）
-w /tmp/ -p w -k tmp_access
-w /var/tmp/ -p w -k var_tmp_access
-w /dev/shm/ -p w -k shm_access

## Web应用文件
-w /var/www/html/ -p wa -k web_content
-w /opt/application/ -p wa -k app_content

# 特权用户操作审计
## root用户所有操作
-a always,exit -F arch=b64 -F auid=0 -S all -k root_actions

## sudo使用审计
-a always,exit -F arch=b64 -S execve -C uid!=euid -F euid=0 -k sudo_use
-a always,exit -F arch=b64 -S execve -C uid!=euid -k setuid_exec

## 用户登录会话
-w /var/log/lastlog -p wa -k lastlog
-w /var/run/utmp -p wa -k utmp
-w /var/log/wtmp -p wa -k wtmp
-w /var/log/btmp -p wa -k btmp

## 密码策略变更
-a always,exit -F arch=b64 -S chmod -F a2&04000 -k setuid_changes
-a always,exit -F arch=b64 -S chown -F success=1 -k ownership_changes

# 网络连接监控
## 原始套接字访问
-a always,exit -F arch=b64 -S socket -F a0=2 -k raw_socket

## 网络接口配置
-a always,exit -F arch=b64 -S ioctl -F a1=0x8912 -k netconfig
-a always,exit -F arch=b64 -S ioctl -F a1=0x8913 -k netconfig
-a always,exit -F arch=b64 -S ioctl -F a1=0x8927 -k netconfig

## 防火墙变更
-w /sbin/iptables -p x -k firewall
-w /sbin/ip6tables -p x -k firewall
-w /sbin/ebtables -p x -k firewall

## DNS查询监控
-a always,exit -F arch=b64 -S socketcall -F a0=1 -k dns_queries

# 审计缓冲区优化
## 根据系统内存调整缓冲区大小
## 小型系统（2-4GB内存）
-b 1024
--backlog_wait_time 0

## 中型系统（8-16GB内存）  
-b 2048
--backlog_wait_time 500

## 大型系统（32GB+内存）
-b 4096
--backlog_wait_time 1000

# 速率限制配置
## 防止审计风暴
-r 1000
## 每秒最大事件数

# 磁盘I/O优化
## 异步刷新配置
flush = INCREMENTAL_ASYNC
freq = 100

# 规则优化技巧
## 使用排除规则减少噪音
-a never,exit -F dir=/proc -k exclude_proc
-a never,exit -F dir=/sys -k exclude_sys
-a never,exit -F dir=/var/cache -k exclude_cache

#!/usr/bin/env python3
"""
自定义审计事件处理器示例
实时分析审计日志并触发告警
"""

import asyncio
import json
import socket
from datetime import datetime

class AuditEventHandler:
    def __init__(self, config_path='/etc/audit/handler.conf'):
        self.config = self.load_config(config_path)
        self.suspicious_patterns = [
            'failed_login',
            'privilege_escalation',
            'sensitive_file_access'
        ]
        
    async def process_audit_event(self, event_data):
        """处理单个审计事件"""
        try:
            event = json.loads(event_data)
            
            # 实时威胁检测
            threat_level = self.assess_threat_level(event)
            if threat_level > self.config['alert_threshold']:
                await self.trigger_alert(event, threat_level)
                
            # 事件关联分析
            await self.correlate_events(event)
            
            # 存储到分析数据库
            await self.store_for_analysis(event)
            
        except Exception as e:
            self.log_error(f"事件处理错误: {str(e)}")
    
    async def correlate_events(self, event):
        """事件关联分析"""
        # 基于时间窗口的关联
        # 基于用户行为的关联
        # 基于源IP的关联
        pass
        
    def assess_threat_level(self, event):
        """评估事件威胁等级"""
        score = 0
        
        # 基于事件类型评分
        event_scores = {
            'USER_LOGIN': 1,
            'USER_ACCT': 3,
            'EXECVE': 2,
            'FILE_ACCESS': 4
        }
        
        score += event_scores.get(event.get('type'), 1)
        
        # 基于用户权限评分
        if event.get('uid') == 0:
            score += 5
            
        # 基于文件敏感度评分
        sensitive_files = ['/etc/shadow', '/etc/sudoers']
        if any(f in event.get('path', '') for f in sensitive_files):
            score += 10
            
        return score

# 使用示例
handler = AuditEventHandler()

#!/bin/bash
# 实时审计日志监控脚本

AUDIT_LOG="/var/log/audit/audit.log"
ALERT_THRESHOLD=5
ALERT_EMAIL="security@company.com"

# 监控函数
monitor_audit_log() {
    tail -F "$AUDIT_LOG" | while read line; do
        # 解析审计记录
        local event_type=$(echo "$line" | grep -o 'type=[^ ]*' | cut -d= -f2)
        local auid=$(echo "$line" | grep -o 'auid=[^ ]*' | cut -d= -f2)
        local success=$(echo "$line" | grep -o 'success=[^ ]*' | cut -d= -f2)
        
        # 关键事件检测
        case "$event_type" in
            "USER_LOGIN")
                if [ "$success" = "no" ]; then
                    handle_failed_login "$line"
                fi
                ;;
            "EXECVE")
                handle_process_execution "$line"
                ;;
            "FILE_ACCESS")
                handle_sensitive_file_access "$line"
                ;;
        esac
    done
}

handle_failed_login() {
    local record="$1"
    local user=$(echo "$record" | grep -o 'acct=[^ ]*' | cut -d= -f2)
    local host=$(echo "$record" | grep -o 'hostname=[^ ]*' | cut -d= -f2)
    
    # 更新失败计数器
    local count=$((++FAILED_LOGINS["$user@$host"]))
    
    if [ $count -ge $ALERT_THRESHOLD ]; then
        send_alert "暴力破解攻击检测" "用户 $user 从 $host 登录失败 $count 次"
    fi
}

# 关联分析函数
correlate_events() {
    ausearch -k suspicious_activity -ts today | \
    aureport -f -i | \
    awk '{
        # 基于时间、用户、源IP进行关联
        print "事件关联:", $0
    }'
}

# 启动监控
declare -A FAILED_LOGINS
monitor_audit_log

import pandas as pd
import numpy as np
from elasticsearch import Elasticsearch
from datetime import datetime, timedelta

class AuditLogAnalyzer:
    def __init__(self, es_host='localhost:9200'):
        self.es = Elasticsearch([es_host])
        self.index_pattern = "audit-*"
    
    def behavioral_analysis(self, user_id, days=30):
        """用户行为分析"""
        # 获取用户历史行为数据
        query = {
            "query": {
                "bool": {
                    "must": [
                        {"term": {"auid": user_id}},
                        {"range": {"@timestamp": {"gte": f"now-{days}d/d"}}}
                    ]
                }
            },
            "aggs": {
                "hourly_pattern": {
                    "date_histogram": {
                        "field": "@timestamp",
                        "calendar_interval": "hour"
                    }
                },
                "command_frequency": {
                    "terms": {"field": "exe.keyword"}
                }
            }
        }
        
        response = self.es.search(index=self.index_pattern, body=query)
        return self.analyze_behavior_patterns(response)
    
    def anomaly_detection(self):
        """异常行为检测"""
        # 基于机器学习的异常检测
        features = self.extract_behavioral_features()
        
        # 使用孤立森林算法检测异常
        from sklearn.ensemble import IsolationForest
        clf = IsolationForest(contamination=0.1)
        anomalies = clf.fit_predict(features)
        
        return anomalies
    
    def timeline_reconstruction(self, start_time, end_time):
        """时间线重建 - 安全事件重建"""
        query = {
            "query": {
                "range": {
                    "@timestamp": {
                        "gte": start_time,
                        "lte": end_time
                    }
                }
            },
            "sort": [{"@timestamp": "asc"}]
        }
        
        events = self.es.search(index=self.index_pattern, body=query, size=10000)
        return self.build_attack_timeline(events)

# 使用示例
analyzer = AuditLogAnalyzer()
user_behavior = analyzer.behavioral_analysis(1000)
anomalies = analyzer.anomaly_detection()

# audisp-remote.conf 远程审计配置
remote_server = 192.168.1.100
port = 60
transport = tcp
queue_depth = 2048
format = managed
network_retry_time = 1
max_tries_per_record = 3
heartbeat_timeout = 60
network_failure_action = stop

# Filebeat 配置 (/etc/filebeat/filebeat.yml)
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/audit/audit.log
  fields:
    type: audit
  json.keys_under_root: true
  json.add_error_key: true

output.elasticsearch:
  hosts: ["elasticsearch:9200"]
  index: "audit-%{+yyyy.MM.dd}"

setup.template:
  name: "audit"
  pattern: "audit-*"

# Logstash 审计管道
input {
  beats {
    port => 5044
  }
}

filter {
  if [type] == "audit" {
    grok {
      match => { "message" => "type=%{WORD:audit_type} msg=audit\(%{NUMBER:timestamp}:%{NUMBER:serial}\):%{GREEDYDATA:audit_data}" }
    }
    
    # 解析审计数据字段
    kv {
      source => "audit_data"
      field_split => " "
      value_split => "="
      target => "audit_fields"
    }
    
    date {
      match => [ "timestamp", "UNIX_MS" ]
      target => "@timestamp"
    }
  }
}

output {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "audit-%{+YYYY.MM.dd}"
  }
}

# PCI DSS专用审计规则
## 持卡人数据环境监控
-w /var/www/html/payment/ -p rwxa -k cardholder_data
-w /opt/application/database/ -p rwxa -k cardholder_data

## 访问控制审计
-a always,exit -F arch=b64 -S open -F success=0 -k pci_access_denied
-a always,exit -F arch=b64 -S truncate -F success=0 -k pci_access_denied

## 加密密钥管理
-w /etc/ssl/private/ -p rwxa -k ssl_keys
-w /opt/application/keys/ -p rwxa -k app_keys

# PCI DSS合规报告生成
#!/bin/bash
generate_pci_report() {
    local start_date=$1
    local end_date=$2
    
    echo "PCI DSS合规审计报告"
    echo "时间范围: $start_date 到 $end_date"
    echo "=========================================="
    
    # 1. 认证尝试
    echo "1. 用户认证活动:"
    aureport -au -i -ts $start_date -te $end_date
    
    # 2. 文件访问
    echo "2. 敏感文件访问:"
    aureport -f -i -ts $start_date -te $end_date | grep -E "(shadow|passwd|sudoers)"
    
    # 3. 特权命令
    echo "3. 特权命令执行:"
    aureport -x -i -ts $start_date -te $end_date
    
    # 4. 网络连接
    echo "4. 网络连接活动:"
    aureport -n -i -ts $start_date -te $end_date
}

# GDPR个人数据保护审计规则
## 个人数据目录监控
-w /var/www/html/user_profiles/ -p rwxa -k gdpr_personal_data
-w /opt/application/customer_data/ -p rwxa -k gdpr_personal_data
-w /var/lib/mysql/customer_db/ -p rwxa -k gdpr_personal_data

## 数据访问审计
-a always,exit -F arch=b64 -S open -F dir=/gdpr_data -k gdpr_access
-a always,exit -F arch=b64 -S read -F dir=/gdpr_data -k gdpr_read
-a always,exit -F arch=b64 -S write -F dir=/gdpr_data -k gdpr_write

## 数据导出监控
-a always,exit -F arch=b64 -S sendto -F success=1 -k data_export
-a always,exit -F arch=b64 -S write -F path=/mnt/external -k external_write

#!/bin/bash
# 审计系统性能监控脚本

monitor_audit_performance() {
    while true; do
        # 监控审计队列状态
        local backlog=$(auditctl -s | grep backlog | awk '{print $2}')
        local lost=$(auditctl -s | grep lost | awk '{print $2}')
        
        # 性能阈值
        local BACKLOG_THRESHOLD=64
        local LOST_THRESHOLD=10
        
        if [ $backlog -gt $BACKLOG_THRESHOLD ]; then
            echo "警告: 审计积压过高 - $backlog" | logger -t audit_monitor
        fi
        
        if [ $lost -gt $LOST_THRESHOLD ]; then
            echo "紧急: 审计事件丢失 - $lost" | logger -t audit_monitor
            # 自动调整缓冲区
            adjust_audit_buffer
        fi
        
        # 磁盘空间检查
        check_audit_disk_space
        
        sleep 60
    done
}

adjust_audit_buffer() {
    # 动态调整审计缓冲区
    local current_buffer=$(auditctl -s | grep "backlog limit" | awk '{print $4}')
    local new_buffer=$((current_buffer * 2))
    
    if [ $new_buffer -le 8192 ]; then
        auditctl -b $new_buffer
        echo "审计缓冲区调整为: $new_buffer" | logger -t audit_monitor
    fi
}

# 审计服务状态检查清单
audit_service_check() {
    echo "=== 审计服务健康检查 ==="
    
    # 1. 服务状态
    systemctl status auditd --no-pager -l
    
    # 2. 规则加载状态
    auditctl -l
    
    # 3. 内核审计状态
    auditctl -s
    
    # 4. 磁盘空间检查
    df -h /var/log/audit/
    
    # 5. 日志文件完整性
    ausearch --input /var/log/audit/audit.log --check
    
    # 6. 最近错误检查
    aureport --start today --errors
    
    # 7. 性能指标
    echo "积压事件: $(auditctl -s | grep backlog | awk '{print $2}')"
    echo "丢失事件: $(auditctl -s | grep lost | awk '{print $2}')"
}

# 自动修复常见问题
audit_auto_fix() {
    # 重启审计服务
    systemctl restart auditd
    sleep 5
    
    # 重新加载规则
    if [ -f /etc/audit/rules.d/audit.rules ]; then
        auditctl -R /etc/audit/rules.d/audit.rules
    fi
    
    # 清理旧日志文件
    find /var/log/audit/ -name "audit.log.*" -mtime +30 -delete
}

# Dockerfile 审计配置
FROM ubuntu:20.04

# 安装审计工具
RUN apt-get update && apt-get install -y auditd audispd-plugins

# 配置容器内审计
COPY audit.rules /etc/audit/rules.d/audit.rules
COPY auditd.conf /etc/audit/auditd.conf

# 启动审计服务
CMD ["/bin/bash", "-c", "service auditd start && tail -f /var/log/audit/audit.log"]

# Kubernetes审计策略
apiVersion: audit.k8s.io/v1
kind: Policy
rules:
  # 记录所有pod创建删除
  - level: Metadata
    resources:
    - group: ""
      resources: ["pods"]
    verbs: ["create", "delete"]
  
  # 记录配置映射访问
  - level: RequestResponse
    resources:
    - group: ""
      resources: ["configmaps"]
  
  # 记录命名空间操作
  - level: Metadata
    resources:
    - group: ""
      resources: ["namespaces"]
    verbs: ["create", "delete", "update"]

#!/usr/bin/env python3
"""
自动化审计管理系统
支持规则部署、监控、报告生成
"""

import asyncio
import yaml
from typing import Dict, List
from dataclasses import dataclass

@dataclass
class AuditRule:
    name: str
    rule: str
    enabled: bool = True
    priority: int = 100

class AuditAutomation:
    def __init__(self, config_file: str):
        self.config = self.load_config(config_file)
        self.rules_manager = AuditRulesManager()
        self.monitor = AuditMonitor()
        self.reporter = ReportGenerator()
    
    async def deploy_rules(self, hosts: List[str], rules: List[AuditRule]):
        """批量部署审计规则"""
        tasks = []
        for host in hosts:
            for rule in rules:
                task = asyncio.create_task(
                    self.deploy_rule_to_host(host, rule)
                )
                tasks.append(task)
        
        await asyncio.gather(*tasks, return_exceptions=True)
    
    async def continuous_monitoring(self):
        """持续监控审计系统状态"""
        while True:
            # 检查审计服务状态
            service_status = await self.monitor.check_service_health()
            
            # 检查规则完整性
            rule_integrity = await self.monitor.verify_rules()
            
            # 生成健康报告
            report = await self.reporter.generate_health_report(
                service_status, rule_integrity
            )
            
            # 发送告警
            if not service_status['healthy']:
                await self.send_alert(service_status)
            
            await asyncio.sleep(300)  # 5分钟间隔

# 使用示例
automation = AuditAutomation('config.yaml')
asyncio.run(automation.continuous_monitoring())

class SmartAlertSystem:
    def __init__(self):
        self.alert_rules = self.load_alert_rules()
        self.alert_history = []
    
    def evaluate_alert(self, event):
        """智能告警评估"""
        alert_score = 0
        triggered_rules = []
        
        for rule in self.alert_rules:
            if self.matches_rule(event, rule):
                alert_score += rule['severity']
                triggered_rules.append(rule['name'])
        
        # 基于历史行为调整分数
        historical_context = self.get_historical_context(event)
        alert_score *= historical_context['risk_multiplier']
        
        if alert_score >= self.alert_threshold:
            self.send_alert(event, alert_score, triggered_rules)
    
    def get_historical_context(self, event):
        """获取历史上下文"""
        user = event.get('auid')
        source_ip = event.get('src')
        
        # 计算用户行为基线偏差
        baseline_deviation = self.calculate_baseline_deviation(user, event)
        
        return {
            'risk_multiplier': max(1.0, baseline_deviation),
            'similar_events_count': self.count_similar_events(event)
        }