Introduction
In today's digital landscape, cyber resilience isn't just about preventing attacks—it's about being prepared to recover quickly when incidents occur. This tutorial will teach you how to implement a recovery readiness framework using Python and cloud infrastructure monitoring tools. You'll learn to build automated recovery checks, monitor system health, and create actionable recovery plans that align with modern resilience standards.
Prerequisites
- Basic understanding of Python programming
- Familiarity with cloud services (AWS/Azure/GCP)
- Access to a cloud platform with monitoring capabilities
- Python 3.7+ installed on your system
- Required Python packages: boto3, requests, pandas, matplotlib
Step-by-step instructions
Step 1: Set Up Your Cloud Monitoring Environment
Install Required Dependencies
First, create a virtual environment and install the necessary packages:
python -m venv recovery_env
source recovery_env/bin/activate # On Windows: recovery_env\Scripts\activate
pip install boto3 requests pandas matplotlib
Why this step is important: Creating a virtual environment isolates your project dependencies and prevents conflicts with other Python projects. This ensures consistent behavior across different systems.
Step 2: Configure Cloud Service Access
Create AWS Configuration File
Create a file called aws_config.py to store your cloud credentials:
import boto3
# Initialize clients
ec2_client = boto3.client('ec2')
cloudwatch_client = boto3.client('cloudwatch')
ssm_client = boto3.client('ssm')
# Define monitoring parameters
class RecoveryConfig:
def __init__(self):
self.region = 'us-east-1'
self.instance_ids = ['i-1234567890abcdef0'] # Replace with your instance IDs
self.metric_names = ['CPUUtilization', 'DiskReadOps', 'NetworkIn']
self.thresholds = {'CPUUtilization': 80, 'DiskReadOps': 1000, 'NetworkIn': 1000000}
Why this step is important: Proper credential management is crucial for security. This approach keeps your credentials separate from your code while making them easily accessible to your monitoring scripts.
Step 3: Implement System Health Monitoring
Create Health Check Functions
Create a file called health_monitor.py to implement system monitoring:
import boto3
import json
from datetime import datetime, timedelta
def get_instance_health(instance_id):
'''Check basic instance health metrics'''
try:
# Get instance status
response = ec2_client.describe_instance_status(
InstanceIds=[instance_id],
IncludeAllInstances=True
)
status = response['InstanceStatuses'][0]
return {
'instance_id': instance_id,
'instance_status': status['InstanceStatus']['Status'],
'system_status': status['SystemStatus']['Status'],
'timestamp': datetime.now().isoformat()
}
except Exception as e:
return {'error': str(e)}
def get_metric_data(instance_id, metric_name, period=300):
'''Fetch recent metric data for recovery readiness'''
try:
end_time = datetime.utcnow()
start_time = end_time - timedelta(minutes=10)
response = cloudwatch_client.get_metric_statistics(
Namespace='AWS/EC2',
MetricName=metric_name,
Dimensions=[{'Name': 'InstanceId', 'Value': instance_id}],
StartTime=start_time,
EndTime=end_time,
Period=period,
Statistics=['Average']
)
if response['Datapoints']:
latest_value = response['Datapoints'][-1]['Average']
return latest_value
return None
except Exception as e:
print(f"Error fetching {metric_name}: {e}")
return None
Why this step is important: This monitoring system evaluates both instance-level and metric-level health indicators, which are critical for determining if a system is ready for recovery operations.
Step 4: Build Automated Recovery Readiness Checks
Create Recovery Readiness Assessment
Create recovery_assessment.py to evaluate system readiness:
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime
class RecoveryReadinessChecker:
def __init__(self, config):
self.config = config
def check_recovery_readiness(self, instance_id):
'''Comprehensive recovery readiness check'''
readiness_score = 0
issues = []
# Check instance health
health = get_instance_health(instance_id)
if health.get('error'):
issues.append(f"Health check failed: {health['error']}")
else:
if health['instance_status'] != 'ok':
issues.append(f"Instance status not OK: {health['instance_status']}")
if health['system_status'] != 'ok':
issues.append(f"System status not OK: {health['system_status']}")
# Check critical metrics
for metric_name, threshold in self.config.thresholds.items():
value = get_metric_data(instance_id, metric_name)
if value and value > threshold:
issues.append(f"{metric_name} above threshold: {value}")
elif value:
readiness_score += 1 # Metric within acceptable range
# Calculate overall score
total_metrics = len(self.config.thresholds)
score_percentage = (readiness_score / total_metrics) * 100
return {
'instance_id': instance_id,
'score': score_percentage,
'issues': issues,
'timestamp': datetime.now().isoformat()
}
def generate_report(self, results):
'''Generate a recovery readiness report'''
df = pd.DataFrame(results)
# Create summary statistics
print("\n=== Recovery Readiness Report ===")
print(f"Total Systems: {len(results)}")
print(f"Average Score: {df['score'].mean():.2f}%")
print(f"Systems Ready: {len(df[df['score'] >= 80])}")
# Generate visual report
plt.figure(figsize=(10, 6))
plt.bar(range(len(df)), df['score'])
plt.xlabel('Instance ID')
plt.ylabel('Recovery Readiness Score')
plt.title('Recovery Readiness by Instance')
plt.xticks(range(len(df)), df['instance_id'], rotation=45)
plt.tight_layout()
plt.savefig('recovery_readiness_report.png')
print("\nReport saved as recovery_readiness_report.png")
Why this step is important: This assessment system provides quantitative metrics for recovery readiness, enabling data-driven decisions about system recovery priorities and readiness levels.
Step 5: Implement Automated Recovery Plans
Create Recovery Action Engine
Create recovery_engine.py to automate recovery actions:
import time
from datetime import datetime
class RecoveryEngine:
def __init__(self, config):
self.config = config
def execute_recovery_plan(self, instance_id, issues):
'''Execute appropriate recovery actions based on issues'''
print(f"\n=== Executing Recovery Plan for {instance_id} ===")
# Log recovery action
with open('recovery_log.txt', 'a') as f:
f.write(f"{datetime.now()}: Recovery initiated for {instance_id}\n")
for issue in issues:
f.write(f" - {issue}\n")
# Basic recovery actions
if any('CPUUtilization' in issue for issue in issues):
print("Warning: High CPU detected. Consider scaling up.")
# In production, this would trigger auto-scaling policies
if any('NetworkIn' in issue for issue in issues):
print("Warning: Network throughput issues detected.")
# Implement network optimization measures
# Restart instance if critical issues found
if len(issues) > 2:
print("Critical issues detected. Initiating instance restart...")
self.restart_instance(instance_id)
print("Recovery plan execution completed.")
def restart_instance(self, instance_id):
'''Restart the specified instance'''
try:
response = ec2_client.reboot_instances(InstanceIds=[instance_id])
print(f"Instance {instance_id} reboot initiated successfully.")
time.sleep(30) # Wait for reboot to complete
except Exception as e:
print(f"Failed to restart instance: {e}")
Why this step is important: Automated recovery actions ensure that systems can respond to incidents without manual intervention, reducing downtime and improving overall resilience.
Step 6: Run Complete Recovery Readiness System
Integrate All Components
Create main.py to orchestrate the entire recovery system:
from aws_config import RecoveryConfig
from health_monitor import get_instance_health, get_metric_data
from recovery_assessment import RecoveryReadinessChecker
from recovery_engine import RecoveryEngine
# Initialize components
config = RecoveryConfig()
checker = RecoveryReadinessChecker(config)
engine = RecoveryEngine(config)
# Main execution loop
if __name__ == '__main__':
results = []
print("Starting Recovery Readiness Assessment...")
for instance_id in config.instance_ids:
# Check recovery readiness
result = checker.check_recovery_readiness(instance_id)
results.append(result)
print(f"{instance_id}: {result['score']:.1f}% ready")
if result['issues']:
print(f"Issues found: {len(result['issues'])}")
for issue in result['issues']:
print(f" - {issue}")
# Execute recovery if needed
if result['score'] < 80 and result['issues']:
engine.execute_recovery_plan(instance_id, result['issues'])
# Generate comprehensive report
checker.generate_report(results)
print("\nRecovery readiness assessment completed.")
Why this step is important: This integration brings all components together to create a complete recovery readiness monitoring system that can automatically assess, report, and act on recovery readiness.
Summary
In this tutorial, you've built a comprehensive recovery readiness framework that monitors cloud infrastructure health, evaluates system readiness, and executes automated recovery actions. This system demonstrates the practical implementation of cyber resilience standards by focusing on the critical aspect of recovery readiness rather than just prevention. The framework can be extended with additional monitoring metrics, more sophisticated recovery actions, and integration with incident response systems.
The key takeaway is that modern cyber resilience requires proactive monitoring and automated recovery capabilities. This approach ensures that systems are not only protected from attacks but are also prepared to recover quickly when incidents occur, making it the new standard for digital resilience.



