How to Lessons Ransomware: Lessons Learned

import boto3
import hashlib
import os
import sys
import json
import logging
from datetime import datetime, timedelta
from botocore.exceptions import ClientError, NoCredentialsError

# Configure logging to stdout for containerized environments
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class ImmutableBackupValidator:
    """Validates air-gapped immutable backups stored in S3-compatible storage with Object Lock enabled."""

    def __init__(self, bucket_name: str, aws_region: str = "us-east-1"):
        self.bucket_name = bucket_name
        self.aws_region = aws_region
        self.s3_client = None
        self.validation_results = {
            "total_backups": 0,
            "valid_backups": 0,
            "invalid_backups": 0,
            "errors": []
        }

        try:
            # Initialize S3 client with explicit credentials check
            self.s3_client = boto3.client("s3", region_name=self.aws_region)
            # Test connectivity by checking bucket existence
            self.s3_client.head_bucket(Bucket=self.bucket_name)
            logger.info(f"Successfully connected to bucket {self.bucket_name}")
        except NoCredentialsError:
            logger.error("AWS credentials not found. Set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY.")
            sys.exit(1)
        except ClientError as e:
            if e.response["Error"]["Code"] == "404":
                logger.error(f"Bucket {self.bucket_name} does not exist")
            else:
                logger.error(f"AWS ClientError: {e}")
            sys.exit(1)

    def _calculate_local_hash(self, file_path: str, algorithm: str = "sha256") -> str:
        """Calculate hash of a local file for integrity comparison."""
        try:
            hash_obj = hashlib.new(algorithm)
            with open(file_path, "rb") as f:
                # Read in chunks to handle large backup files
                for chunk in iter(lambda: f.read(4096), b""):
                    hash_obj.update(chunk)
            return hash_obj.hexdigest()
        except FileNotFoundError:
            logger.error(f"Local file {file_path} not found")
            return ""
        except PermissionError:
            logger.error(f"Permission denied reading {file_path}")
            return ""

    def validate_backup(self, backup_key: str, local_checksum_path: str = None) -> bool:
        """Validate a single backup object: check Object Lock status, checksum, retention period."""
        try:
            # 1. Check if Object Lock is enabled for the object
            object_lock = self.s3_client.get_object_retention(Bucket=self.bucket_name, Key=backup_key)
            retention_until = datetime.fromisoformat(object_lock["Retention"]["RetainUntilDate"].isoformat())
            if retention_until < datetime.now(retention_until.tzinfo):
                logger.warning(f"Backup {backup_key} has expired retention: {retention_until}")
                return False

            # 2. Verify checksum if provided
            if local_checksum_path:
                local_hash = self._calculate_local_hash(local_checksum_path)
                if not local_hash:
                    return False
                # Get S3 stored checksum (assumes checksum is stored as object metadata)
                s3_metadata = self.s3_client.head_object(Bucket=self.bucket_name, Key=backup_key)
                s3_hash = s3_metadata.get("Metadata", {}).get("sha256-checksum", "")
                if not s3_hash:
                    logger.error(f"No SHA256 checksum metadata found for {backup_key}")
                    return False
                if local_hash != s3_hash:
                    logger.error(f"Checksum mismatch for {backup_key}: local {local_hash} vs S3 {s3_hash}")
                    return False

            # 3. Check backup is not modified (S3 Object Lock prevents overwrite, but verify ETag)
            s3_etag = self.s3_client.head_object(Bucket=self.bucket_name, Key=backup_key)["ETag"]
            logger.info(f"Backup {backup_key} validated: retention until {retention_until}, ETag {s3_etag}")
            return True

        except ClientError as e:
            logger.error(f"Failed to validate {backup_key}: {e}")
            return False

    def run_batch_validation(self, prefix: str = "backups/") -> dict:
        """Validate all backups under a given S3 prefix."""
        try:
            paginator = self.s3_client.get_paginator("list_objects_v2")
            for page in paginator.paginate(Bucket=self.bucket_name, Prefix=prefix):
                if "Contents" not in page:
                    logger.warning(f"No backups found under prefix {prefix}")
                    break
                for obj in page["Contents"]:
                    backup_key = obj["Key"]
                    self.validation_results["total_backups"] += 1
                    # Assume checksum file is same key with .sha256 suffix
                    checksum_key = f"{backup_key}.sha256"
                    # Check if checksum file exists in S3
                    try:
                        self.s3_client.head_object(Bucket=self.bucket_name, Key=checksum_key)
                        # Download checksum file temporarily
                        local_checksum = f"/tmp/{os.path.basename(checksum_key)}"
                        self.s3_client.download_file(self.bucket_name, checksum_key, local_checksum)
                    except ClientError:
                        local_checksum = None

                    if self.validate_backup(backup_key, local_checksum):
                        self.validation_results["valid_backups"] += 1
                    else:
                        self.validation_results["invalid_backups"] += 1

                    # Clean up temporary checksum file
                    if local_checksum and os.path.exists(local_checksum):
                        os.remove(local_checksum)

            logger.info(f"Batch validation complete: {json.dumps(self.validation_results)}")
            return self.validation_results

        except ClientError as e:
            logger.error(f"Failed to list backups: {e}")
            return self.validation_results

if __name__ == "__main__":
    # Example usage: validate backups in prod-immutable-backups bucket
    validator = ImmutableBackupValidator(bucket_name="prod-immutable-backups", aws_region="us-west-2")
    results = validator.run_batch_validation(prefix="daily/2024-05-")
    if results["invalid_backups"] > 0:
        logger.error(f"Found {results['invalid_backups']} invalid backups")
        sys.exit(1)
    sys.exit(0)

// SPDX-License-Identifier: GPL-2.0
#include 
#include 
#include 
#include 
#include 
#include 
#include 

// Define a BPF map to track file write rates per process
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 1024);
    __type(key, pid_t);
    __type(value, u64); // Number of file writes in 1-second window
    __uint(pinning, LIBBPF_PIN_BY_NAME);
} file_write_count SEC(".maps");

// Define a BPF map to store alert state (avoid spamming)
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 1024);
    __type(key, pid_t);
    __type(value, u64); // Last alert timestamp
    __uint(pinning, LIBBPF_PIN_BY_NAME);
} last_alert SEC(".maps");

// Threshold: 100 file writes per second triggers alert
#define WRITE_THRESHOLD 100
#define ALERT_COOLDOWN_NS 5000000000 // 5 seconds between alerts per PID

// Tracepoint for file write (vfs_write)
SEC("tracepoint/syscalls/sys_enter_write")
int sys_enter_write(struct trace_event_raw_sys_enter *ctx) {
    pid_t pid = bpf_get_current_pid_tgid() >> 32;
    u64 *count = bpf_map_lookup_elem(&file_write_count, &pid);
    u64 new_count = 1;

    if (count) {
        new_count = *count + 1;
    }
    bpf_map_update_elem(&file_write_count, &pid, &new_count, BPF_ANY);

    // Check if threshold exceeded
    if (new_count >= WRITE_THRESHOLD) {
        // Check cooldown to avoid alert spam
        u64 now = bpf_ktime_get_ns();
        u64 *last_alert_time = bpf_map_lookup_elem(&last_alert, &pid);
        if (last_alert_time && (now - *last_alert_time) < ALERT_COOLDOWN_NS) {
            return 0;
        }

        // Get process name for alert
        char comm[16];
        bpf_get_current_comm(comm, sizeof(comm));

        // Log alert to trace pipe
        bpf_printk("RANSOMWARE_ALERT: PID %d (%s) wrote %llu files in 1s window", pid, comm, new_count);

        // Update last alert time
        bpf_map_update_elem(&last_alert, &pid, &now, BPF_ANY);

        // TODO: Add logic to kill process or send event to userspace
    }

    return 0;
}

// Tracepoint for file rename (common ransomware tactic: rename to .locked)
SEC("tracepoint/syscalls/sys_enter_rename")
int sys_enter_rename(struct trace_event_raw_sys_enter *ctx) {
    pid_t pid = bpf_get_current_pid_tgid() >> 32;
    char oldpath[256] = {0};
    char newpath[256] = {0};

    // Read old and new path arguments from syscall
    bpf_probe_read_user_str(oldpath, sizeof(oldpath), (const char *)ctx->args[0]);
    bpf_probe_read_user_str(newpath, sizeof(newpath), (const char *)ctx->args[1]);

    // Check if new path ends with common ransomware extensions
    char *ransom_ext[] = {".locked", ".encrypted", ".ryuk", ".conti", ".lockbit"};
    int num_ext = sizeof(ransom_ext) / sizeof(ransom_ext[0]);

    for (int i = 0; i < num_ext; i++) {
        int ext_len = bpf_strnlen(ransom_ext[i], 32);
        int path_len = bpf_strnlen(newpath, 256);
        if (path_len < ext_len) continue;
        // Compare last ext_len characters of newpath with ransom_ext[i]
        if (bpf_memcmp(newpath + path_len - ext_len, ransom_ext[i], ext_len) == 0) {
            char comm[16];
            bpf_get_current_comm(comm, sizeof(comm));
            bpf_printk("RANSOMWARE_ALERT: PID %d (%s) renamed %s to %s (ransomware extension detected)", pid, comm, oldpath, newpath);
            // Update alert cooldown
            u64 now = bpf_ktime_get_ns();
            bpf_map_update_elem(&last_alert, &pid, &now, BPF_ANY);
            break;
        }
    }

    return 0;
}

// Reset write count every 1 second using a BPF timer (Linux 5.15+)
struct {
    __uint(type, BPF_MAP_TYPE_TIMER);
    __uint(max_entries, 1);
} reset_timer SEC(".maps");

int reset_write_counts(struct bpf_map *map, void *key, void *value, void *ctx) {
    // Delete all entries in file_write_count map
    pid_t pid = 0;
    u64 *val;
    while ((val = bpf_map_get_next_key(&file_write_count, &pid, &pid))) {
        bpf_map_delete_elem(&file_write_count, &pid);
    }
    // Reschedule timer for 1 second later
    bpf_timer_start(&reset_timer, 1000000000, 0);
    return 0;
}

SEC("tp_btf/timer")
int BPF_PROG(timer_handler, struct bpf_timer *timer) {
    bpf_timer_cancel(timer);
    bpf_map_for_each_elem(&file_write_count, reset_write_counts, NULL, 0);
    return 0;
}

char _license[] SEC("license") = "GPL";

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/ec2"
    "github.com/slack-go/slack"
    "golang.org/x/crypto/ssh"
)

// IRConfig holds configuration for incident response actions
type IRConfig struct {
    SlackToken      string `json:"slack_token"`
    SlackChannel    string `json:"slack_channel"`
    AWSRegion       string `json:"aws_region"`
    JumpHost        string `json:"jump_host"`
    JumpUser        string `json:"jump_user"`
    JumpKeyPath     string `json:"jump_key_path"`
    ContainmentSGID string `json:"containment_sg_id"` // Security group that blocks all inbound/outbound traffic
}

// RansomwareAlert represents an incoming alert from detection tools
type RansomwareAlert struct {
    AlertID    string    `json:"alert_id"`
    Timestamp  time.Time `json:"timestamp"`
    HostID     string    `json:"host_id"` // EC2 instance ID or hostname
    HostIP     string    `json:"host_ip"`
    PID        int       `json:"pid"`
    Process    string    `json:"process"`
    Severity   string    `json:"severity"` // "high", "critical"
}

// IncidentResponder handles automated containment and notification
type IncidentResponder struct {
    config      IRConfig
    sess        *session.Session
    ec2Client   *ec2.EC2
    slackClient *slack.Client
}

// NewIncidentResponder initializes a new responder with config validation
func NewIncidentResponder(configPath string) (*IncidentResponder, error) {
    // Read and parse config file
    configFile, err := os.ReadFile(configPath)
    if err != nil {
        return nil, fmt.Errorf("failed to read config: %w", err)
    }
    var config IRConfig
    if err := json.Unmarshal(configFile, &config); err != nil {
        return nil, fmt.Errorf("failed to parse config: %w", err)
    }

    // Validate required config fields
    if config.SlackToken == "" || config.AWSRegion == "" || config.ContainmentSGID == "" {
        return nil, fmt.Errorf("missing required config fields")
    }

    // Initialize AWS session
    sess, err := session.NewSession(&aws.Config{
        Region: aws.String(config.AWSRegion),
    })
    if err != nil {
        return nil, fmt.Errorf("failed to create AWS session: %w", err)
    }

    // Initialize Slack client
    slackClient := slack.New(config.SlackToken)

    return &IncidentResponder{
        config:      config,
        sess:        sess,
        ec2Client:   ec2.New(sess),
        slackClient: slackClient,
    }, nil
}

// IsolateHost attaches a containment security group to the EC2 instance to block all traffic
func (ir *IncidentResponder) IsolateHost(ctx context.Context, instanceID string) error {
    // Get current security groups for the instance
    describeInput := &ec2.DescribeInstancesInput{
        InstanceIds: aws.StringSlice([]string{instanceID}),
    }
    result, err := ir.ec2Client.DescribeInstances(describeInput)
    if err != nil {
        return fmt.Errorf("failed to describe instance %s: %w", instanceID, err)
    }
    if len(result.Reservations) == 0 || len(result.Reservations[0].Instances) == 0 {
        return fmt.Errorf("instance %s not found", instanceID)
    }

    instance := result.Reservations[0].Instances[0]
    // Preserve existing security groups, add containment SG
    existingSGs := instance.SecurityGroups
    newSGs := append(existingSGs, &ec2.GroupIdentifier{GroupId: aws.String(ir.config.ContainmentSGID)})

    // Modify instance security groups
    _, err = ir.ec2Client.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
        InstanceId: aws.String(instanceID),
        Groups:     newSGs,
    })
    if err != nil {
        return fmt.Errorf("failed to modify security groups for %s: %w", instanceID, err)
    }

    log.Printf("Successfully isolated instance %s with containment SG %s", instanceID, ir.config.ContainmentSGID)
    return nil
}

// NotifySOC sends a high-priority alert to the SOC Slack channel
func (ir *IncidentResponder) NotifySOC(alert RansomwareAlert) error {
    message := fmt.Sprintf(
        "🚨 *RANSOMWARE ALERT* 🚨\n"+
            "Alert ID: %s\n"+
            "Host: %s (%s)\n"+
            "Process: %s (PID %d)\n"+
            "Severity: %s\n"+
            "Timestamp: %s\n"+
            "Host has been isolated. Begin forensic analysis immediately.",
        alert.AlertID, alert.HostID, alert.HostIP, alert.Process, alert.PID, alert.Severity, alert.Timestamp.Format(time.RFC3339),
    )

    _, _, err := ir.slackClient.PostMessage(
        ir.config.SlackChannel,
        slack.MsgOptionText(message, false),
        slack.MsgOptionAttachments(slack.Attachment{
            Color: "danger",
            Fields: []slack.AttachmentField{
                {Title: "Alert ID", Value: alert.AlertID, Short: true},
                {Title: "Host ID", Value: alert.HostID, Short: true},
            },
        }),
    )
    if err != nil {
        return fmt.Errorf("failed to send Slack notification: %w", err)
    }

    log.Printf("Sent SOC notification for alert %s", alert.AlertID)
    return nil
}

// CaptureMemory uses SSH to a jump host to capture memory dump of the compromised instance
func (ir *IncidentResponder) CaptureMemory(instanceIP string) error {
    // Read SSH key for jump host
    key, err := os.ReadFile(ir.config.JumpKeyPath)
    if err != nil {
        return fmt.Errorf("failed to read SSH key: %w", err)
    }
    signer, err := ssh.ParsePrivateKey(key)
    if err != nil {
        return fmt.Errorf("failed to parse SSH key: %w", err)
    }

    // Connect to jump host
    config := &ssh.ClientConfig{
        User: ir.config.JumpUser,
        Auth: []ssh.AuthMethod{
            ssh.PublicKeys(signer),
        },
        HostKeyCallback: ssh.InsecureIgnoreHostKey(), // TODO: Use known hosts in production
    }
    client, err := ssh.Dial("tcp", ir.config.JumpHost+":22", config)
    if err != nil {
        return fmt.Errorf("failed to connect to jump host: %w", err)
    }
    defer client.Close()

    // Execute memory dump command (assumes LiME is installed on target)
    session, err := client.NewSession()
    if err != nil {
        return fmt.Errorf("failed to create SSH session: %w", err)
    }
    defer session.Close()

    // Command to dump memory to /tmp/memdump.lime (truncated for brevity)
    cmd := fmt.Sprintf("ssh -o StrictHostKeyChecking=no ec2-user@%s 'sudo lime-forensics /tmp/memdump.lime'", instanceIP)
    output, err := session.CombinedOutput(cmd)
    if err != nil {
        return fmt.Errorf("failed to capture memory: %w, output: %s", err, output)
    }

    log.Printf("Memory dump captured for %s: %s", instanceIP, output)
    return nil
}

func main() {
    // Load configuration
    responder, err := NewIncidentResponder("ir-config.json")
    if err != nil {
        log.Fatalf("Failed to initialize responder: %v", err)
    }

    // Example: Process a mock ransomware alert
    alert := RansomwareAlert{
        AlertID:   "alert-2024-05-20-001",
        Timestamp: time.Now(),
        HostID:    "i-0123456789abcdef0",
        HostIP:    "10.0.1.123",
        PID:       1234,
        Process:   "encryptor",
        Severity:  "critical",
    }

    ctx := context.Background()

    // Step 1: Isolate the compromised host
    if err := responder.IsolateHost(ctx, alert.HostID); err != nil {
        log.Printf("Failed to isolate host: %v", err)
    }

    // Step 2: Notify SOC
    if err := responder.NotifySOC(alert); err != nil {
        log.Printf("Failed to notify SOC: %v", err)
    }

    // Step 3: Capture memory dump for forensics
    if err := responder.CaptureMemory(alert.HostIP); err != nil {
        log.Printf("Failed to capture memory: %v", err)
    }

    log.Printf("Incident response completed for alert %s", alert.AlertID)
}

package main

import (
    "fmt"
    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/s3"
)

func enableS3ObjectLock(bucketName string) error {
    sess := session.Must(session.NewSession(&aws.Config{Region: aws.String("us-east-1")}))
    svc := s3.New(sess)

    // Enable Object Lock on the bucket (requires versioning to be enabled first)
    _, err := svc.PutObjectLockConfiguration(&s3.PutObjectLockConfigurationInput{
        Bucket: aws.String(bucketName),
        ObjectLockConfiguration: &s3.ObjectLockConfiguration{
            ObjectLockEnabled: aws.String("Enabled"),
            Rule: &s3.ObjectLockRule{
                DefaultRetention: &s3.DefaultRetention{
                    Mode: aws.String("COMPLIANCE"),
                    Days: aws.Int64(30),
                },
            },
        },
    })
    if err != nil {
        return fmt.Errorf("failed to enable Object Lock: %w", err)
    }
    return nil
}

apiVersion: cilium.io/v1alpha1
kind: TracingPolicy
metadata:
  name: ransomware-extension-detection
spec:
  kprobes:
  - call: "sys_rename"
    syscall: true
    args:
    - index: 0
      type: "string"
    - index: 1
      type: "string"
    selectors:
    - matchArgs:
      - index: 1
        operator: "EndsWith"
        values:
        - ".locked"
        - ".encrypted"
        - ".ryuk"
        - ".conti"
        - ".lockbit"
    action: "Sigkill"
    message: "Ransomware extension rename detected: {{.args[0]}} -> {{.args[1]}}"

SELECT
  process.pid,
  process.name,
  process.path,
  file.operation,
  file.path,
  count(*) as write_count
FROM
  process
JOIN
  file_events ON process.pid = file_events.pid
WHERE
  file.operation = 'write'
  AND time > (SELECT unix_time - 60 FROM time)
GROUP BY
  process.pid
HAVING
  write_count > 100;

ransomware-lessons/
├── backup_validator.py       # Code Example 1: Immutable Backup Validator
├── ebpf_detector.c           # Code Example 2: eBPF Ransomware Detection
├── incident_responder.go     # Code Example 3: Automated Incident Response
├── benchmark_results/        # Raw benchmark data from 42 enterprise environments
│   ├── backup_recovery.csv
│   ├── ebpf_performance.csv
│   └── ir_timing.csv
├── deploy/                   # Deployment manifests for Kubernetes
│   ├── tetragon-policy.yaml
│   └── ir-config.json
└── README.md                 # Setup and usage instructions