1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

Ransomware Alert Triggered
          │
          ▼
  Is this a true positive?
  (EDR confidence ≥ HIGH or
   ransom note confirmed?)
          │
     YES  │  NO
          │   └──► Escalate to analyst for review
          ▼
  How many endpoints affected?
          │
    ┌─────┴──────┐
  1–3            4+
  hosts          hosts
    │              │
    ▼              ▼
Isolate         Initiate
silently        Crisis Bridge
                Notify CISO
                    │
                    ▼
         Isolate all affected hosts
         Block lateral movement (network segmentation)
                    │
          ┌─────────┴──────────┐
       Backups              Backups
       intact?              compromised?
          │                      │
          ▼                      ▼
    Begin recovery          Engage DR plan
    from clean backup       Notify legal/insurance

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

def validate_ransomware_alert(alert):
    confidence_score = 0

    # EDR detection from a known ransomware family
    if alert.edr_classification in KNOWN_RANSOMWARE_FAMILIES:
        confidence_score += 50

    # Shadow copy deletion — near-certain ransomware indicator
    if alert.process_cmdline and any(cmd in alert.process_cmdline for cmd in [
        'vssadmin delete shadows',
        'wmic shadowcopy delete',
        'bcdedit /set recoveryenabled no'
    ]):
        confidence_score += 40

    # Mass file rename (>100 files with new extension in 60 seconds)
    if alert.file_rename_count > 100:
        confidence_score += 30

    # Ransom note file created
    if alert.file_created and any(
        note in alert.file_created.lower()
        for note in ['readme', 'decrypt', 'how_to', 'ransom', 'restore_files']
    ):
        confidence_score += 30

    return confidence_score >= 50, confidence_score

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

def isolate_endpoints(affected_hosts, edr_client):
    isolation_results = []

    for host in affected_hosts:
        try:
            # CrowdStrike network containment
            result = edr_client.contain_host(device_id=host.device_id)
            isolation_results.append({
                'hostname': host.hostname,
                'status': 'ISOLATED',
                'timestamp': datetime.utcnow().isoformat(),
                'method': 'EDR network containment'
            })
        except Exception as e:
            # Fallback: firewall-level block
            firewall_block_host(host.ip_address)
            isolation_results.append({
                'hostname': host.hostname,
                'status': 'ISOLATED_FALLBACK',
                'method': 'Firewall ACL'
            })

    return isolation_results

def assess_blast_radius(initial_host, siem_client, lookback_minutes=30):
    """Find other hosts that communicated with the affected host
       in the window before detection."""
    query = f"""
    index=network_traffic
    (src_ip="{initial_host.ip}" OR dest_ip="{initial_host.ip}")
    earliest=-{lookback_minutes}m
    | stats values(src_ip) as sources, values(dest_ip) as destinations by _time
    """
    results = siem_client.search(query)
    lateral_candidates = set()
    for r in results:
        lateral_candidates.update(r['sources'] + r['destinations'])
    lateral_candidates.discard(initial_host.ip)
    return list(lateral_candidates)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

import hashlib
import requests

def identify_ransomware_family(ransom_note_text, encrypted_file_extension):
    """
    Query ID Ransomware (public API) to identify the family
    from the ransom note or encrypted file extension.
    """
    # Check extension against known families database
    known_extensions = {
        '.lockbit': 'LockBit',
        '.blackcat': 'BlackCat/ALPHV',
        '.hive':    'Hive',
        '.phobos':  'Phobos',
        '.dharma':  'Dharma/CrySis',
        '.ryuk':    'Ryuk',
    }
    family = known_extensions.get(encrypted_file_extension.lower(), 'Unknown')

    # Cross-reference with VirusTotal if we have a sample hash
    return {
        'family': family,
        'extension': encrypted_file_extension,
        'note_preview': ransom_note_text[:200] if ransom_note_text else None
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

def verify_backup_integrity(affected_systems, backup_client):
    """
    Check whether backups for affected systems exist and
    were last written BEFORE the ransomware activity window.
    """
    backup_status = []

    for system in affected_systems:
        backups = backup_client.get_backups(system.hostname)

        if not backups:
            backup_status.append({
                'system': system.hostname,
                'status': 'NO_BACKUP',
                'action': 'ESCALATE_TO_DR'
            })
            continue

        latest_backup = max(backups, key=lambda b: b.timestamp)
        attack_start = system.first_detection_time

        if latest_backup.timestamp < attack_start:
            backup_status.append({
                'system': system.hostname,
                'status': 'CLEAN_BACKUP_AVAILABLE',
                'backup_time': latest_backup.timestamp.isoformat(),
                'action': 'PROCEED_WITH_RECOVERY'
            })
        else:
            backup_status.append({
                'system': system.hostname,
                'status': 'BACKUP_POTENTIALLY_ENCRYPTED',
                'action': 'MANUAL_VERIFICATION_REQUIRED'
            })

    return backup_status

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

def notify_stakeholders(incident, scope, family_info, backup_status):
    severity = 'P1' if scope['hosts_affected'] >= 4 else 'P2'

    # SOC channel
    teams_post_to_channel(
        channel='#soc-alerts',
        title=f"[{severity}] RANSOMWARE INCIDENT — {incident.id}",
        body=f"""
**Family:** {family_info['family']}
**Hosts isolated:** {len(scope['isolated_hosts'])}
**Backup status:** {sum(1 for b in backup_status if b['status'] == 'CLEAN_BACKUP_AVAILABLE')} clean / {len(backup_status)} total
**Extension:** {family_info['extension']}
        """
    )

    # For P1, also notify CISO and IT manager via email
    if severity == 'P1':
        send_email(
            to=['ciso@company.ae', 'it-manager@company.ae'],
            subject=f"[URGENT] Active Ransomware Incident — {scope['hosts_affected']} systems affected",
            body=build_executive_summary(incident, scope, family_info)
        )

    # Create P1 incident ticket
    ticket = servicenow_create_incident(
        priority=1,
        category='Ransomware',
        short_desc=f"Ransomware: {family_info['family']} — {scope['hosts_affected']} hosts",
        description=build_full_report(incident, scope, family_info, backup_status)
    )

    return ticket

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

def block_ransomware_iocs(family_info, firewall_client, proxy_client):
    """
    Pull known C2 IPs and domains for the identified family from MISP
    and block them at the perimeter immediately.
    """
    if family_info['family'] == 'Unknown':
        return []

    iocs = misp_search_by_tag(f'ransomware:{family_info["family"].lower()}')
    blocked = []

    for ioc in iocs:
        if ioc['type'] == 'ip-dst':
            firewall_client.block_ip(ioc['value'], comment=f"Ransomware C2 — {family_info['family']}")
            blocked.append(ioc['value'])
        elif ioc['type'] == 'domain':
            proxy_client.block_domain(ioc['value'])
            blocked.append(ioc['value'])

    return blocked

1
2
3
4
5
6
7
8
9

□ Verify all affected hosts are isolated from the network
□ Confirm backup integrity for each affected system
□ Rebuild affected systems from clean image (do not decrypt in place)
□ Restore data from last clean backup
□ Reset credentials for all accounts active on affected systems
□ Rotate service account passwords for systems in blast radius
□ Patch the initial access vector before reconnecting to network
□ Enable additional monitoring on rebuilt systems for 30 days
□ Conduct post-incident review within 72 hours