监控数据完整性校验的脚本如何写

wen 实用脚本 2026-07-04 172

本文目录导读：

监控数据完整性校验的脚本如何写

基本哈希校验脚本 (Linux/Unix)
Python版本 (更强大的校验)
配置文件示例
使用 systemd 服务
定时任务设置
使用说明

我来提供一个完整的监控数据完整性校验脚本模板,包含多种校验方式：

基本哈希校验脚本 (Linux/Unix)

#!/bin/bash
# 监控数据完整性校验脚本
# 功能：计算文件或目录的哈希值，并与基准值对比
# 配置
MONITOR_DIR="/path/to/monitor"    # 监控目录
BASELINE_FILE="/path/to/baseline.md5"  # 基准文件
LOG_FILE="/var/log/integrity_check.log"
ALERT_EMAIL="admin@example.com"
# 颜色输出
RED='\033[0;31m'
GREEN='\033[0;32m'
NC='\033[0m'
# 日志函数
log() {
    echo "$(date '+%Y-%m-%d %H:%M:%S') - $1" >> "$LOG_FILE"
    echo -e "$2$1${NC}"
}
# 检查文件是否存在
check_file_exists() {
    if [ ! -f "$1" ]; then
        log "错误: 文件 $1 不存在" "$RED"
        return 1
    fi
    return 0
}
# 创建基准哈希文件
create_baseline() {
    log "创建基准哈希文件..." "$GREEN"
    find "$MONITOR_DIR" -type f -exec md5sum {} \; > "$BASELINE_FILE"
    log "基准文件已创建: $BASELINE_FILE" "$GREEN"
}
# 校验数据完整性
check_integrity() {
    local changed_files=0
    local missing_files=0
    local new_files=0
    log "开始数据完整性校验..." "$GREEN"
    # 检查基准文件是否存在
    if [ ! -f "$BASELINE_FILE" ]; then
        log "基准文件不存在，正在创建..." "$RED"
        create_baseline
        return
    fi
    # 读取基准文件并校验
    while IFS= read -r line; do
        expected_hash=$(echo "$line" | cut -d' ' -f1)
        file_path=$(echo "$line" | cut -d' ' -f2-)
        if [ -f "$file_path" ]; then
            current_hash=$(md5sum "$file_path" | cut -d' ' -f1)
            if [ "$expected_hash" != "$current_hash" ]; then
                log "文件已修改: $file_path" "$RED"
                changed_files=$((changed_files + 1))
                # 发送告警
                send_alert "文件已被修改: $file_path"
            fi
        else
            log "文件丢失: $file_path" "$RED"
            missing_files=$((missing_files + 1))
            send_alert "文件丢失: $file_path"
        fi
    done < "$BASELINE_FILE"
    # 检查新增文件
    find "$MONITOR_DIR" -type f | while read -r file; do
        if ! grep -q "$file" "$BASELINE_FILE" 2>/dev/null; then
            log "新增文件: $file" "$RED"
            new_files=$((new_files + 1))
            send_alert "新增未授权文件: $file"
        fi
    done
    # 输出结果
    log "校验完成:" "$GREEN"
    log "  修改文件数: $changed_files" "$GREEN"
    log "  丢失文件数: $missing_files" "$GREEN"
    log "  新增文件数: $new_files" "$GREEN"
    return $changed_files
}
# 发送告警
send_alert() {
    local message="$1"
    echo "[数据完整性告警] $message" | mail -s "数据完整性告警" "$ALERT_EMAIL"
}
# 定期检查函数
scheduled_check() {
    while true; do
        check_integrity
        sleep 3600  # 每小时检查一次
    done
}
# 恢复文件 (从备份)
restore_file() {
    local file="$1"
    local backup_dir="/path/to/backup"
    if [ -f "$backup_dir/$(basename $file)" ]; then
        cp "$backup_dir/$(basename $file)" "$file"
        log "文件已恢复: $file" "$GREEN"
    else
        log "无法恢复: $file (备份不存在)" "$RED"
    fi
}
# 主函数
main() {
    case "$1" in
        create)
            create_baseline
            ;;
        check)
            check_integrity
            ;;
        watch)
            scheduled_check
            ;;
        restore)
            if [ -z "$2" ]; then
                echo "用法: $0 restore <文件名>"
                exit 1
            fi
            restore_file "$2"
            ;;
        *)
            echo "用法: $0 {create|check|watch|restore}"
            echo "  create  - 创建基准哈希文件"
            echo "  check   - 执行一次完整性检查"
            echo "  watch   - 持续监控 (每小时检查)"
            echo "  restore - 恢复文件"
            exit 1
            ;;
    esac
}
main "$@"

Python版本 (更强大的校验)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import sys
import json
import hashlib
import logging
import smtplib
from datetime import datetime
from email.mime.text import MIMEText
from pathlib import Path
class IntegrityMonitor:
    """数据完整性监控类"""
    def __init__(self, config):
        self.config = config
        self.setup_logging()
    def setup_logging(self):
        """配置日志系统"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(self.config['log_file']),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def calculate_hash(self, file_path, algorithm='sha256', chunk_size=8192):
        """计算文件哈希值"""
        try:
            hash_obj = hashlib.new(algorithm)
            with open(file_path, 'rb') as f:
                while chunk := f.read(chunk_size):
                    hash_obj.update(chunk)
            return hash_obj.hexdigest()
        except Exception as e:
            self.logger.error(f"计算哈希失败 {file_path}: {e}")
            return None
    def get_file_info(self, file_path):
        """获取文件信息"""
        try:
            stat = os.stat(file_path)
            return {
                'path': str(file_path),
                'size': stat.st_size,
                'modified_time': datetime.fromtimestamp(stat.st_mtime).isoformat(),
                'created_time': datetime.fromtimestamp(stat.st_ctime).isoformat(),
                'permissions': oct(stat.st_mode)[-3:],
                'owner': stat.st_uid,
                'group': stat.st_gid
            }
        except Exception as e:
            self.logger.error(f"获取文件信息失败 {file_path}: {e}")
            return None
    def create_baseline(self, directory):
        """创建基准文件"""
        baseline = {}
        # 支持的文件扩展名
        valid_extensions = self.config.get('valid_extensions', None)
        for file_path in Path(directory).rglob('*'):
            if file_path.is_file():
                # 检查文件类型
                if valid_extensions:
                    if file_path.suffix.lower() not in valid_extensions:
                        continue
                file_str = str(file_path)
                hash_value = self.calculate_hash(file_path)
                file_info = self.get_file_info(file_path)
                if hash_value and file_info:
                    baseline[file_str] = {
                        'hash': hash_value,
                        'info': file_info,
                        'algorithm': self.config.get('hash_algorithm', 'sha256')
                    }
        # 保存基准文件
        with open(self.config['baseline_file'], 'w') as f:
            json.dump(baseline, f, indent=2)
        self.logger.info(f"已创建基准文件: {self.config['baseline_file']}")
        return baseline
    def load_baseline(self):
        """加载基准文件"""
        try:
            with open(self.config['baseline_file'], 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            self.logger.error("基准文件不存在")
            return None
        except json.JSONDecodeError:
            self.logger.error("基准文件格式错误")
            return None
    def check_integrity(self, directory):
        """检查数据完整性"""
        baseline = self.load_baseline()
        if not baseline:
            self.logger.warning("未找到基准文件，正在创建...")
            baseline = self.create_baseline(directory)
            return
        violations = {
            'modified': [],
            'missing': [],
            'new': []
        }
        current_files = set()
        # 检查现有文件
        for file_path in Path(directory).rglob('*'):
            if file_path.is_file():
                file_str = str(file_path)
                current_files.add(file_str)
                if file_str in baseline:
                    # 检查哈希值
                    expected_hash = baseline[file_str]['hash']
                    current_hash = self.calculate_hash(file_path)
                    if current_hash and current_hash != expected_hash:
                        violations['modified'].append({
                            'file': file_str,
                            'expected': expected_hash,
                            'current': current_hash,
                            'old_info': baseline[file_str].get('info', {}),
                            'new_info': self.get_file_info(file_path)
                        })
                else:
                    violations['new'].append({
                        'file': file_str,
                        'info': self.get_file_info(file_path)
                    })
        # 检查丢失的文件
        for file_str in baseline:
            if file_str not in current_files:
                violations['missing'].append({
                    'file': file_str,
                    'last_known': baseline[file_str].get('info', {})
                })
        # 记录违规信息
        self.log_violations(violations)
        # 发送告警
        if any(violations.values()):
            self.send_alert(violations)
        return violations
    def log_violations(self, violations):
        """记录违规信息"""
        if violations['modified']:
            self.logger.warning(f"发现修改的文件: {len(violations['modified'])}")
            for v in violations['modified']:
                self.logger.warning(f"  修改: {v['file']}")
        if violations['missing']:
            self.logger.warning(f"发现丢失的文件: {len(violations['missing'])}")
            for v in violations['missing']:
                self.logger.warning(f"  丢失: {v['file']}")
        if violations['new']:
            self.logger.warning(f"发现新增文件: {len(violations['new'])}")
            for v in violations['new']:
                self.logger.warning(f"  新增: {v['file']}")
    def send_alert(self, violations):
        """发送告警"""
        if not self.config.get('email_alerts'):
            return
        subject = f"[完整性告警] {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
        body = "数据完整性违规报告\n"
        body += "=" * 50 + "\n\n"
        if violations['modified']:
            body += f"修改的文件 ({len(violations['modified'])}):\n"
            for v in violations['modified']:
                body += f"  - {v['file']}\n"
        if violations['missing']:
            body += f"\n丢失的文件 ({len(violations['missing'])}):\n"
            for v in violations['missing']:
                body += f"  - {v['file']}\n"
        if violations['new']:
            body += f"\n新增文件 ({len(violations['new'])}):\n"
            for v in violations['new']:
                body += f"  - {v['file']}\n"
        msg = MIMEText(body)
        msg['Subject'] = subject
        msg['From'] = self.config['email_from']
        msg['To'] = self.config['email_to']
        try:
            with smtplib.SMTP(self.config['smtp_server'], self.config['smtp_port']) as server:
                if self.config.get('smtp_tls'):
                    server.starttls()
                if self.config.get('smtp_user'):
                    server.login(self.config['smtp_user'], self.config['smtp_password'])
                server.send_message(msg)
            self.logger.info("告警邮件已发送")
        except Exception as e:
            self.logger.error(f"发送告警邮件失败: {e}")
def main():
    """主函数"""
    config = {
        'baseline_file': '/path/to/baseline.json',
        'log_file': '/var/log/integrity_monitor.log',
        'hash_algorithm': 'sha256',
        'valid_extensions': ['.txt', '.csv', '.conf', '.py', '.sh', '.json'],
        'email_alerts': False,
        'email_from': 'monitor@example.com',
        'email_to': 'admin@example.com',
        'smtp_server': 'smtp.example.com',
        'smtp_port': 587,
        'smtp_tls': True,
        'smtp_user': None,
        'smtp_password': None
    }
    monitor = IntegrityMonitor(config)
    if len(sys.argv) < 2:
        print("用法: python integrity_monitor.py [create|check|watch] [directory]")
        sys.exit(1)
    directory = sys.argv[2] if len(sys.argv) > 2 else '/path/to/monitor'
    if sys.argv[1] == 'create':
        monitor.create_baseline(directory)
    elif sys.argv[1] == 'check':
        violations = monitor.check_integrity(directory)
        if violations:
            print(json.dumps(violations, indent=2))
    elif sys.argv[1] == 'watch':
        import time
        while True:
            monitor.check_integrity(directory)
            time.sleep(3600)  # 每小时检查
    else:
        print("无效命令")
if __name__ == "__main__":
    main()

配置文件示例

# config.yaml
integrity_monitor:
  monitor_directories:
    - path: /data/production
      recursive: true
      file_patterns:
        - "*.txt"
        - "*.csv"
        - "*.conf"
      exclude_patterns:
        - "*.tmp"
        - "*.log"
        - ".git/"
    - path: /etc/important
      recursive: false
      file_patterns: ["*"]
      exclude_patterns: []
  checksum_algorithm: sha256  # sha256, sha512, md5
  baseline:
    file: /var/lib/integrity/baseline.json
    auto_update: false
    update_schedule: "0 0 * * 0"  # 每周日更新
  alerts:
    email:
      enabled: false
      smtp_server: smtp.example.com
      smtp_port: 587
      use_tls: true
      from: monitor@example.com
      to: admin@example.com
    webhook:
      enabled: false
      url: https://hooks.slack.com/services/xxx
    syslog:
      enabled: true
      facility: local0
  logging:
    file: /var/log/integrity_monitor.log
    level: INFO  # DEBUG, INFO, WARNING, ERROR
  actions:
    on_modified: log_and_alert  # log, alert, restore, quarantine
    on_missing: log_and_alert
    on_new: log_and_alert
    backup_dir: /backup/data

使用 systemd 服务

# /etc/systemd/system/integrity-monitor.service
[Unit]
Description=Data Integrity Monitor
After=network.target
[Service]
Type=simple
User=root
ExecStart=/usr/local/bin/integrity_monitor.py watch /data/production
Restart=always
RestartSec=60
[Install]
WantedBy=multi-user.target

定时任务设置

# crontab -e
# 每小时检查一次
0 * * * * /usr/local/bin/integrity_check.sh check
# 每天凌晨2点更新基准
0 2 * * * /usr/local/bin/integrity_check.sh create
# 监控日志
*/30 * * * * /usr/local/bin/integrity_check.sh check >> /var/log/integrity_check.log 2>&1

使用说明

初始化基准：

./integrity_check.sh create
# 或
python integrity_monitor.py create /path/to/monitor

执行校验：

./integrity_check.sh check
# 或
python integrity_monitor.py check /path/to/monitor

持续监控：

./integrity_check.sh watch
# 或
python integrity_monitor.py watch /path/to/monitor

这个脚本可以根据你的具体需求进行定制，比如添加数据库监控、网络存储监控等功能。

本文地址： https://www.bmjup.com/post/15710.html