雲端安全監控與自動化事件回應

import boto3
from datetime import datetime, timedelta

def create_cloudwatch_alarm(instance_id, threshold):
 cloudwatch = boto3.client('cloudwatch')
 alarm_name = f'HighCPUUsage-{instance_id}'
 
 response = cloudwatch.put_metric_alarm(
 AlarmName=alarm_name,
 ComparisonOperator='GreaterThanThreshold',
 EvaluationPeriods=1,
 MetricName='CPUUtilization',
 Namespace='AWS/EC2',
 Period=300,
 Statistic='Average',
 Threshold=threshold,
 ActionsEnabled=True,
 AlarmActions=[
 'arn:aws:sns:REGION:ACCOUNT_ID:TOPIC_NAME'
 ],
 AlarmDescription=f'CPU utilization exceeds {threshold}%',
 Dimensions=[
 {
 'Name': 'InstanceId',
 'Value': instance_id
 }
 ]
 )
 
 return response

# #### 內容解密：
# 此函式建立了一個CloudWatch警示，用於監控指定EC2例項的CPU使用率。當CPU使用率超過設定的閾值時，觸發SNS通知。

import boto3

def create_eventbridge_rule():
 eventbridge = boto3.client('events')
 rule_name = 'TriggerLambdaOnCloudWatchAlarm'
 
 response = eventbridge.put_rule(
 Name=rule_name,
 EventPattern='{"source": ["aws.cloudwatch"]}',
 State='ENABLED'
 )
 
 eventbridge.put_targets(
 Rule=rule_name,
 Targets=[
 {
 'Id': 'LambdaFunction',
 'Arn': 'arn:aws:lambda:REGION:ACCOUNT_ID:function:LAMBDA_FUNCTION_NAME'
 }
 ]
 )
 
 return response

# #### 內容解密：
# 此函式建立了一個EventBridge規則，用於監聽CloudWatch警示事件。當事件觸發時，會呼叫指定的Lambda函式執行自動化回應。

import boto3
import json

def lambda_handler(event, context):
 ec2 = boto3.client('ec2')
 instance_id = event['detail']['instance-id']
 
 try:
 response = ec2.stop_instances(InstanceIds=[instance_id])
 print(f'Stopped instance: {instance_id}')
 
 response = ec2.start_instances(InstanceIds=[instance_id])
 print(f'Started instance: {instance_id}')
 
 return {
 'statusCode': 200,
 'body': json.dumps(f'Restarted instance: {instance_id}')
 }
 except Exception as e:
 print(f'Error restarting instance: {e}')
 return {
 'statusCode': 500,
 'body': json.dumps(f'Error restarting instance: {e}')
 }

# #### 內容解密：
# 此Lambda函式接收事件通知，重啟指定的EC2例項。首先停止例項，然後再啟動，以完成重啟操作。

import pandas as pd
from sklearn.ensemble import IsolationForest

def detect_anomalies(data):
 df = pd.DataFrame(data)
 model = IsolationForest(contamination=0.01)
 df['anomaly'] = model.fit_predict(df)
 
 anomalies = df[df['anomaly'] == -1]
 return anomalies

# #### 內容解密：
# 此函式使用Isolation Forest演算法對輸入資料進行異常檢測。異常資料將被標記為-1，用於識別潛在的安全威脅。

# 事件監控與自動化應對的模擬實作
import time
from datetime import datetime

class EventMonitor:
    def __init__(self):
        # 初始化事件監控系統
        self.events = []

    def monitor(self, event):
        # 監控事件並記錄
        self.events.append({
            'timestamp': datetime.now(),
            'event': event
        })
        print(f"監控到事件：{event}")

class AutoResponder:
    def __init__(self, monitor):
        # 初始化自動化應對系統並繫結事件監控系統
        self.monitor = monitor

    def respond(self):
        # 根據監控到的事件進行自動化應對
        while True:
            if self.monitor.events:
                event = self.monitor.events.pop(0)
                print(f"自動應對事件：{event['event']} 於 {event['timestamp']}")
                # 模擬自動化應對措施
                time.sleep(1)  # 延遲模擬處理時間
            else:
                time.sleep(0.1)  # 無事件時等待

# 建立事件監控系統例項
monitor = EventMonitor()

# 建立自動化應對系統例項
responder = AutoResponder(monitor)

# 模擬事件監控與自動化應對流程
import threading
threading.Thread(target=responder.respond).start()

# 模擬產生事件
for i in range(10):
    monitor.monitor(f"事件 {i+1}")
    time.sleep(0.5)