Day-302-安全自动化脚本实战

# Day 323: 安全自动化脚本实战 - Python 脚本/API 集成/自动化工具

> 安全运维系列第 33 天 | 预计阅读时间：40 分钟 | 难度：★★★★☆

---

## 清单 目录

1. [自动化概述](#自动化概述)
2. [Python 安全脚本](#python-安全脚本)
3. [API 集成实战](#api-集成实战)
4. [日志处理脚本](#日志处理脚本)
5. [自动化工具](#自动化工具)
6. [实战案例](#实战案例)
7. [总结与思考](#总结与思考)
8. [参考资料](#参考资料)

---

## 自动化概述

### 自动化场景

**数据收集**：
```
日志收集:
- 系统日志
- 应用日志
- 安全日志
- 网络日志

威胁情报:
- IOC 收集
- 漏洞信息
- 威胁报告
- 安全公告
```

**数据分析**：
```
日志分析:
- 日志解析
- 模式识别
- 异常检测
- 关联分析

威胁分析:
- IOC 匹配
- 威胁评分
- 攻击链分析
- 影响评估
```

**响应处置**：
```
自动响应:
- IP 封禁
- 账户禁用
- 文件隔离
- 通知发送

报告生成:
- 日报周报
- 事件报告
- 合规报告
- 审计报告
```

### 自动化价值

**效率提升**：
```
时间节省:
- 重复任务自动化
- 7x24 运行
- 并行处理
- 快速响应

质量提升:
- 标准化流程
- 减少人为错误
- 完整记录
- 可追溯
```

**成本降低**：
```
人力成本:
- 减少手动工作
- 优化人员配置
- 提高人效

运营成本:
- 工具整合
- 流程优化
- 资源优化
```

---

## Python 安全脚本

### 基础脚本

**日志解析脚本**：
```python
#!/usr/bin/env python3
"""
日志解析脚本
解析系统日志，提取安全事件
"""

import re
from datetime import datetime

class LogParser:
    def __init__(self, log_file):
        self.log_file = log_file
        self.patterns = {
            'failed_login': r'Failed password for (\S+) from (\S+)',
            'successful_login': r'Accepted password for (\S+) from (\S+)',
            'sudo': r'sudo:\s+(\S+).*COMMAND=(\S+)',
        }
    
    def parse(self):
        events = []
        with open(self.log_file, 'r') as f:
            for line in f:
                event = self.parse_line(line)
                if event:
                    events.append(event)
        return events
    
    def parse_line(self, line):
        for event_type, pattern in self.patterns.items():
            match = re.search(pattern, line)
            if match:
                return {
                    'type': event_type,
                    'timestamp': self.extract_timestamp(line),
                    'details': match.groups(),
                    'raw': line.strip()
                }
        return None
    
    def extract_timestamp(self, line):
        # 提取时间戳逻辑
        return datetime.now().isoformat()

# 使用示例
parser = LogParser('/var/log/auth.log')
events = parser.parse()
for event in events:
    print(f"{event['type']}: {event['details']}")
```

**威胁情报查询脚本**：
```python
#!/usr/bin/env python3
"""
威胁情报查询脚本
查询多个 TI 平台的 IOC 信息
"""

import requests
import hashlib

class ThreatIntel:
    def __init__(self, api_keys):
        self.api_keys = api_keys
        self.base_urls = {
            'virustotal': 'https://www.virustotal.com/api/v3',
            'abuseipdb': 'https://api.abuseipdb.com/api/v2',
        }
    
    def query_ip(self, ip):
        """查询 IP 信誉"""
        results = {}
        
        # VirusTotal
        vt_url = f"{self.base_urls['virustotal']}/ip_addresses/{ip}"
        vt_headers = {'x-apikey': self.api_keys['virustotal']}
        vt_response = requests.get(vt_url, headers=vt_headers)
        results['virustotal'] = vt_response.json()
        
        # AbuseIPDB
        abuse_url = f"{self.base_urls['abuseipdb']}/check"
        abuse_params = {'ipAddress': ip, 'maxAgeInDays': 90}
        abuse_headers = {'Key': self.api_keys['abuseipdb']}
        abuse_response = requests.get(abuse_url, params=abuse_params, headers=abuse_headers)
        results['abuseipdb'] = abuse_response.json()
        
        return results
    
    def query_hash(self, file_hash):
        """查询文件哈希"""
        vt_url = f"{self.base_urls['virustotal']}/files/{file_hash}"
        vt_headers = {'x-apikey': self.api_keys['virustotal']}
        vt_response = requests.get(vt_url, headers=vt_headers)
        return vt_response.json()

# 使用示例
api_keys = {
    'virustotal': 'YOUR_VT_API_KEY',
    'abuseipdb': 'YOUR_ABUSE_API_KEY',
}

ti = ThreatIntel(api_keys)
results = ti.query_ip('1.2.3.4')
print(results)
```

### 高级脚本

**自动化响应脚本**：
```python
#!/usr/bin/env python3
"""
自动化响应脚本
根据告警自动执行响应动作
"""

import requests
import json
from datetime import datetime

class AutoResponder:
    def __init__(self, config):
        self.config = config
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f"Bearer {config['api_token']}",
            'Content-Type': 'application/json'
        })
    
    def block_ip(self, ip, reason):
        """封禁 IP"""
        # 防火墙封禁
        fw_response = self.session.post(
            f"{self.config['firewall_url']}/api/block",
            json={'ip': ip, 'reason': reason, 'duration': '24h'}
        )
        
        # WAF 封禁
        waf_response = self.session.post(
            f"{self.config['waf_url']}/api/blocklist",
            json={'ip': ip, 'reason': reason}
        )
        
        return {
            'firewall': fw_response.status_code == 200,
            'waf': waf_response.status_code == 200
        }
    
    def disable_user(self, username, reason):
        """禁用用户"""
        # AD 禁用
        ad_response = self.session.post(
            f"{self.config['ad_url']}/api/disable",
            json={'username': username, 'reason': reason}
        )
        
        return ad_response.status_code == 200
    
    def send_notification(self, channel, message):
        """发送通知"""
        if channel == 'slack':
            response = self.session.post(
                self.config['slack_webhook'],
                json={'text': message}
            )
        elif channel == 'email':
            response = self.session.post(
                f"{self.config['mail_url']}/api/send",
                json={'to': self.config['alert_email'], 'subject': 'Security Alert', 'body': message}
            )
        
        return response.status_code == 200
    
    def process_alert(self, alert):
        """处理告警"""
        actions = []
        
        if alert['severity'] == 'critical':
            # 封禁攻击 IP
            if 'source_ip' in alert:
                result = self.block_ip(alert['source_ip'], alert['description'])
                actions.append({'action': 'block_ip', 'result': result})
            
            # 禁用被攻陷账户
            if 'compromised_user' in alert:
                result = self.disable_user(alert['compromised_user'], alert['description'])
                actions.append({'action': 'disable_user', 'result': result})
            
            # 发送通知
            message = f"Critical Alert: {alert['description']}"
            self.send_notification('slack', message)
            self.send_notification('email', message)
        
        return {
            'alert_id': alert['id'],
            'actions': actions,
            'timestamp': datetime.now().isoformat()
        }

# 使用示例
config = {
    'api_token': 'YOUR_API_TOKEN',
    'firewall_url': 'https://firewall.example.com',
    'waf_url': 'https://waf.example.com',
    'ad_url': 'https://ad.example.com',
    'slack_webhook': 'https://hooks.slack.com/...',
    'mail_url': 'https://mail.example.com',
    'alert_email': 'soc@example.com'
}

responder = AutoResponder(config)
alert = {
    'id': 'ALERT-001',
    'severity': 'critical',
    'description': 'Brute force attack detected',
    'source_ip': '1.2.3.4',
    'compromised_user': 'admin'
}
result = responder.process_alert(alert)
print(json.dumps(result, indent=2))
```

---

## API 集成实战

### SIEM API 集成

**Splunk API**：
```python
#!/usr/bin/env python3
"""
Splunk API 集成脚本
执行搜索查询，获取告警
"""

import requests
import time
from requests.auth import HTTPBasicAuth

class SplunkClient:
    def __init__(self, config):
        self.base_url = config['url']
        self.auth = HTTPBasicAuth(config['username'], config['password'])
        self.session = requests.Session()
        self.session.auth = self.auth
        self.session.verify = False  # 生产环境应使用有效证书
    
    def search(self, query, earliest='-1h', latest='now'):
        """执行搜索查询"""
        # 创建搜索任务
        create_url = f"{self.base_url}/services/search/jobs"
        create_data = {
            'search': f'search {query}',
            'earliest_time': earliest,
            'latest_time': latest,
            'output_mode': 'json'
        }
        create_response = self.session.post(create_url, data=create_data)
        sid = create_response.json()['sid']
        
        # 等待搜索完成
        while True:
            status_url = f"{self.base_url}/services/search/jobs/{sid}"
            status_response = self.session.get(status_url)
            status = status_response.json()['entry'][0]['content']
            
            if status['isDone']:
                break
            time.sleep(1)
        
        # 获取结果
        results_url = f"{self.base_url}/services/search/jobs/{sid}/results"
        results_params = {'output_mode': 'json', 'count': 1000}
        results_response = self.session.get(results_url, params=results_params)
        
        return results_response.json()['results']
    
    def get_alerts(self):
        """获取告警"""
        query = 'index=security NOT action=allowed | stats count by src_ip, action'
        return self.search(query)

# 使用示例
config = {
    'url': 'https://splunk.example.com:8089',
    'username': 'admin',
    'password': 'YOUR_PASSWORD'
}

splunk = SplunkClient(config)
alerts = splunk.get_alerts()
for alert in alerts:
    print(f"{alert['src_ip']}: {alert['action']} ({alert['count']})")
```

### EDR API 集成

**CrowdStrike API**：
```python
#!/usr/bin/env python3
"""
CrowdStrike EDR API 集成
查询主机状态，执行响应动作
"""

import requests

class CrowdStrikeClient:
    def __init__(self, config):
        self.base_url = config['url']
        self.client_id = config['client_id']
        self.client_secret = config['client_secret']
        self.token = None
        self.session = requests.Session()
    
    def authenticate(self):
        """获取访问令牌"""
        url = f"{self.base_url}/oauth2/token"
        data = {
            'client_id': self.client_id,
            'client_secret': self.client_secret,
            'grant_type': 'client_credentials'
        }
        response = self.session.post(url, data=data)
        self.token = response.json()['access_token']
        self.session.headers.update({'Authorization': f"Bearer {self.token}"})
    
    def get_hosts(self, filter=None):
        """获取主机列表"""
        if not self.token:
            self.authenticate()
        
        url = f"{self.base_url}/devices/queries/devices-scroll/v1"
        params = {'filter': filter} if filter else {}
        response = self.session.get(url, params=params)
        return response.json()
    
    def contain_host(self, host_id):
        """隔离主机"""
        if not self.token:
            self.authenticate()
        
        url = f"{self.base_url}/containment/entities/containment/v1"
        data = {'ids': [host_id], 'action': 'contain'}
        response = self.session.post(url, json=data)
        return response.json()
    
    def run_command(self, host_id, command):
        """执行命令"""
        if not self.token:
            self.authenticate()
        
        url = f"{self.base_url}/real-time-response/entities/queued-commands/v1"
        data = {
            'base_command': 'runscript',
            'host_ids': [host_id],
            'command_string': f"runscript -raw='{command}'"
        }
        response = self.session.post(url, json=data)
        return response.json()

# 使用示例
config = {
    'url': 'https://api.crowdstrike.com',
    'client_id': 'YOUR_CLIENT_ID',
    'client_secret': 'YOUR_CLIENT_SECRET'
}

cs = CrowdStrikeClient(config)
hosts = cs.get_hosts()
print(f"Found {len(hosts['resources'])} hosts")

# 隔离可疑主机
# cs.contain_host('HOST_ID')
```

---

## 日志处理脚本

### 日志收集脚本

```python
#!/usr/bin/env python3
"""
日志收集脚本
从多个源收集日志，发送到 SIEM
"""

import logging
import requests
from logging.handlers import SysLogHandler

class LogCollector:
    def __init__(self, config):
        self.config = config
        self.logger = logging.getLogger('LogCollector')
        self.logger.setLevel(logging.INFO)
        
        # 配置 Syslog 处理器
        syslog_handler = SysLogHandler(
            address=(config['siem_host'], config['siem_port']),
            facility=SysLogHandler.LOG_LOCAL0
        )
        formatter = logging.Formatter(
            '%(asctime)s %(hostname)s %(program)s: %(message)s'
        )
        syslog_handler.setFormatter(formatter)
        self.logger.addHandler(syslog_handler)
    
    def collect_system_logs(self):
        """收集系统日志"""
        with open('/var/log/syslog', 'r') as f:
            for line in f:
                self.logger.info(line.strip(), extra={
                    'hostname': 'server01',
                    'program': 'syslog'
                })
    
    def collect_auth_logs(self):
        """收集认证日志"""
        with open('/var/log/auth.log', 'r') as f:
            for line in f:
                self.logger.info(line.strip(), extra={
                    'hostname': 'server01',
                    'program': 'auth'
                })
    
    def collect_application_logs(self, log_file, program):
        """收集应用日志"""
        with open(log_file, 'r') as f:
            for line in f:
                self.logger.info(line.strip(), extra={
                    'hostname': 'server01',
                    'program': program
                })

# 使用示例
config = {
    'siem_host': 'siem.example.com',
    'siem_port': 514
}

collector = LogCollector(config)
collector.collect_system_logs()
collector.collect_auth_logs()
collector.collect_application_logs('/var/log/nginx/access.log', 'nginx')
```

### 日志分析脚本

```python
#!/usr/bin/env python3
"""
日志分析脚本
分析日志，检测异常
"""

import re
from collections import Counter
from datetime import datetime, timedelta

class LogAnalyzer:
    def __init__(self):
        self.failed_logins = Counter()
        self.successful_logins = Counter()
        self.threshold = 5  # 失败登录阈值
        self.time_window = timedelta(minutes=5)
    
    def analyze_auth_log(self, log_file):
        """分析认证日志"""
        failed_pattern = r'Failed password for (\S+) from (\S+)'
        success_pattern = r'Accepted password for (\S+) from (\S+)'
        
        with open(log_file, 'r') as f:
            for line in f:
                # 失败登录
                failed_match = re.search(failed_pattern, line)
                if failed_match:
                    user, ip = failed_match.groups()
                    self.failed_logins[ip] += 1
                    
                    # 检查是否超过阈值
                    if self.failed_logins[ip] >= self.threshold:
                        print(f"[ALERT] Brute force detected from {ip}")
                
                # 成功登录
                success_match = re.search(success_pattern, line)
                if success_match:
                    user, ip = success_match.groups()
                    self.successful_logins[ip] += 1
                    
                    # 检查失败后成功 (可能暴力破解成功)
                    if self.failed_logins[ip] >= 3:
                        print(f"[ALERT] Possible successful brute force from {ip}")
    
    def generate_report(self):
        """生成分析报告"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'failed_logins': dict(self.failed_logins),
            'successful_logins': dict(self.successful_logins),
            'alerts': []
        }
        
        # 添加告警
        for ip, count in self.failed_logins.items():
            if count >= self.threshold:
                report['alerts'].append({
                    'type': 'brute_force',
                    'source_ip': ip,
                    'count': count
                })
        
        return report

# 使用示例
analyzer = LogAnalyzer()
analyzer.analyze_auth_log('/var/log/auth.log')
report = analyzer.generate_report()
print(report)
```

---

## 自动化工具

### Ansible 安全自动化

**安全基线部署**：
```yaml
# security_baseline.yml
---
- name: Deploy Security Baseline
  hosts: all
  become: yes
  
  tasks:
    - name: Ensure password policy
      lineinfile:
        path: /etc/login.defs
        regexp: "^PASS_MAX_DAYS"
        line: "PASS_MAX_DAYS 90"
    
    - name: Ensure SSH security
      lineinfile:
        path: /etc/ssh/sshd_config
        regexp: "^PermitRootLogin"
        line: "PermitRootLogin no"
    
    - name: Ensure firewall enabled
      systemd:
        name: firewalld
        state: started
        enabled: yes
    
    - name: Install security updates
      yum:
        name: '*'
        state: latest
        security: yes
```

**应急响应剧本**：
```yaml
# incident_response.yml
---
- name: Incident Response
  hosts: "{{ target_host }}"
  become: yes
  
  tasks:
    - name: Isolate host from network
      command: iptables -I INPUT -j DROP
    
    - name: Collect evidence
      archive:
        path:
          - /var/log/
          - /tmp/
        dest: /evidence/evidence_{{ ansible_date_time.iso8601 }}.tar.gz
    
    - name: Send notification
      uri:
        url: "{{ slack_webhook }}"
        method: POST
        body_format: json
        body:
          text: "Host {{ target_host }} isolated due to security incident"
```

### SOAR Playbook

**钓鱼邮件响应 Playbook**：
```yaml
# phishing_response.yml
---
playbook:
  name: "Phishing Email Response"
  version: "1.0"
  
  triggers:
    - alert_type: "phishing_email"
  
  steps:
    - name: "Extract Indicators"
      action: "email_parse"
      params:
        email_id: "${alert.email_id}"
    
    - name: "Threat Intel Lookup"
      parallel:
        - action: "virustotal_lookup"
          params:
            indicators: "${email_data.indicators}"
        - action: "urlhaus_lookup"
          params:
            urls: "${email_data.urls}"
    
    - name: "Risk Assessment"
      action: "risk_scoring"
      params:
        ti_score: "${ti_result.score}"
    
    - name: "Decision"
      type: "switch"
      on: "${risk_score}"
      cases:
        - condition: ">= 80"
          playbook: "high_risk_response"
        - condition: ">= 50"
          playbook: "medium_risk_response"
        - default:
          playbook: "low_risk_response"
```

---

## 实战案例

### 案例 1: 自动化威胁情报收集

**项目概况**：
```
目标：自动化收集威胁情报
数据源：10+ TI 平台
输出：标准化 IOC 格式
频率：每小时
```

**实施方案**：
```python
#!/usr/bin/env python3
"""
自动化威胁情报收集脚本
"""

import requests
import json
from datetime import datetime

class TICollector:
    def __init__(self, config):
        self.config = config
        self.iocs = {'ips': [], 'domains': [], 'hashes': []}
    
    def collect_from_virustotal(self):
        """从 VirusTotal 收集"""
        # 实现逻辑
        pass
    
    def collect_from_abuseipdb(self):
        """从 AbuseIPDB 收集"""
        # 实现逻辑
        pass
    
    def normalize_ioc(self, ioc):
        """标准化 IOC 格式"""
        return {
            'type': ioc['type'],
            'value': ioc['value'],
            'confidence': ioc['confidence'],
            'source': ioc['source'],
            'timestamp': datetime.now().isoformat()
        }
    
    def save_to_siem(self):
        """保存到 SIEM"""
        # 实现逻辑
        pass

# 定时执行
# cron: 0 * * * * /usr/bin/python3 /opt/scripts/ti_collector.py
```

### 案例 2: 自动化事件响应

**项目概况**：
```
目标：自动化常见事件响应
场景：暴力破解、恶意软件、数据泄露
响应时间：< 5 分钟
```

**实施方案**：
```
触发条件:
- SIEM 告警
- EDR 检测
- 用户报告

自动响应:
- IP 封禁
- 账户禁用
- 文件隔离
- 通知发送

人工确认:
- 高风险决策
- 不可逆操作
- 升级处置
```

---

## 总结与思考

### 核心要点回顾

1. **自动化场景**：数据收集、数据分析、响应处置
2. **Python 脚本**：日志解析、威胁情报、自动响应
3. **API 集成**：SIEM API、EDR API、防火墙 API
4. **日志处理**：日志收集、日志分析、异常检测
5. **自动化工具**：Ansible、SOAR Playbook
6. **实战案例**：TI 收集、事件响应

### 深入思考

**自动化挑战**：
- 工具集成复杂
- 误报处理困难
- 维护成本高
- 技能要求高

**发展方向**：
- 低代码平台
- AI 驱动自动化
- 云原生自动化
- 标准化接口

### 最佳实践

**开发建议**：
- 代码版本控制
- 完整文档
- 错误处理
- 日志记录

**运营建议**：
- 定期测试
- 性能监控
- 持续优化
- 知识管理

---

## 参考资料

### 学习资源

- **Python Security Programming**: https://nostarch.com/pythonsecurity
- **Automating Security Operations**: SANS 课程
- **SOAR Best Practices**: Gartner 报告

### 工具资源

- **Requests**: https://requests.readthedocs.io
- **Ansible**: https://www.ansible.com
- **Cortex XSOAR**: https://www.paloaltonetworks.com/cortex/xsoar

---

*Day 323 完成 | 安全自动化脚本实战详解 | 字数：约 20,000 字 (新增)*

*安全运维系列 33 篇完整教程完结！完成*