Refactor scenarios

.gitignore

@@ -55,8 +55,16 @@ terraform.tfstate.backup
 
 *aws-scenario-1-output.json
 *aws-scenario-2-output.json
+*aws-scenario-*
+*gcp-scenario*
 *token.txt
 *cobra-as1-report.html
 *cobra-as2-report.html
 *cnbas-as1-report.html
-*cnbas-as2-report.html
+*cnbas-as2-report.html
+
+
+files/var/logs/*
+files/var/output/*
+files/var/reports/*
+!.gitkeep

README.md

@@ -1,7 +1,7 @@
 <h2 align="center">🚀 Cloud Offensive Breach and Risk Assessment (COBRA) Tool 👩‍💻</h2>
 
 <!-- <p align="center">
-<img width="396" alt="cobra" src="https://github.com/PaloAltoNetworks/cobra-tool/assets/4271325/f618c9c8-4f3f-48ca-848b-c51b53e4e366">
+<img width="300" alt="cobra" src="https://raw.githubusercontent.com/PaloAltoNetworks/cobra-tool/main/core/cobra-logo.png">
 </p> -->
 
 [![License](https://img.shields.io/badge/License-Apache_2.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)
@@ -37,8 +37,13 @@ It facilitates Proof of Concept (POC) evaluations, assesses security controls, m
 
 - Python 3.8+
 - pip3
-- Pulumi Account 
-- AWS CLI
+- Pulumi CLI [Docs](https://www.pulumi.com/docs/install/)
+- Pulumi Account [here](https://www.pulumi.com/)
+  - Create Pulumi Personal Access Token [Docs](https://www.pulumi.com/docs/pulumi-cloud/access-management/access-tokens/#creating-personal-access-tokens)
+  - Use shell to login to Pulumi `$pulumi login` (Paste access token) [Docs](https://www.pulumi.com/docs/cli/commands/pulumi_login/)
+- AWS CLI installed
+  - Will use the default profile credentials unless defined with the environment variables `AWS_PROFILE` and `AWS_REGION`
+  - Must have the region defined. 
 - Azure CLI
 - Google Cloud SDK
 
@@ -104,6 +109,8 @@ options:
                         Scenario selection
 ```
 
+>>Note* ONLY RUN INTO SANDBOX ENVIRONMENT
+
 #### Simulate AWS Scenario 
 
 ```
@@ -112,16 +119,19 @@ python3 cobra.py aws launch --simulation
 
 
 ```
-  ____ ___  ____  ____      _
- / ___/ _ \| __ )|  _ \    / \
-| |  | | | |  _ \| |_) |  / _ \
-| |__| |_| | |_) |  _ <  / ___ \
- \____\___/|____/|_| \_\/_/   \_\
+  ____    ___    ____    ____       _
+ / ___|  / _ \  | __ )  |  _ \     / \
+| |     | | | | |  _ \  | |_) |   / _ \
+| |___  | |_| | | |_) | |  _ <   / ___ \
+ \____|  \___/  |____/  |_| \_\ /_/   \_\
 
 
 Select Attack Scenario of aws:
 1. Exploit Vulnerable Application, EC2 takeover, Credential Exfiltration & Anomalous Compute Provisioning
 2. Rest API exploit - command injection, credential exfiltration from backend lambda and privilige escalation, rogue identity creation & persistence
+3. Compromising a web app living inside a GKE Pod, access pod secret, escalate privilege, take over the cluster
+4. Exfiltrate EC2 role credentials using IMDSv2 with least privileged access
+5. Instance takeover, abuse s3 access & perform ransomware using external KMS key
 Enter your choice:
 
 ```
@@ -142,13 +152,16 @@ python3 cobra.py aws destroy --scenario <scenario-1/scenario-2>
 
 1. Exploit Vulnerable Application, EC2 takeover, Credential Exfiltration & Anomalous Compute Provisioning
 2. Rest API exploit - command injection, credential exfiltration from backend lambda and privilige escalation, rogue identity creation & persistence
+3. Compromising a web app living inside a GKE Pod, access pod secret, escalate privilege, take over the cluster
+4. Exfiltrate EC2 role credentials using IMDSv2 with least privileged access
+5. Instance takeover, abuse s3 access & perform ransomware using external KMS key
 
 ### To Do / In Roadmap
 
-3. Compromising a GKE Pod and accessing cluster secrets, taking over the cluster & escalating privileges at the Project level, possible project takeover. 
-4. Azure App exploit on a function, data exfiltration from Blob storage & abusing function misconfigs to escalate privileges & leaving a backdoor IAM entity. 
-5. Exploiting an App on VM, exfiltration of data from Cosmos DB & possible takeover of a resource group. 
-6. More scenarios loading...
+
+- Azure App exploit on a function, data exfiltration from Blob storage & abusing function misconfigs to escalate privileges & leaving a backdoor IAM entity. 
+- Exploiting an App on VM, exfiltration of data from Cosmos DB & possible takeover of a resource group. 
+- More scenarios loading...
 
 ## License
 

cobra.py

@@ -1,30 +1,24 @@
 import argparse
+
 from core import main
-import pyfiglet
-from termcolor import colored
+from helpers.main import get_scenario_list
 
-def display_banner():
-    ascii_art = pyfiglet.figlet_format("COBRA")
-    print(colored(ascii_art, color="cyan"))
 
 def parse_arguments():
+    scenarios = get_scenario_list()
     parser = argparse.ArgumentParser(description="Terminal-based option tool")
     parser.add_argument("cloud_provider", choices=["aws", "azure", "gcp"], help="Cloud provider (aws, azure, gcp)")
     parser.add_argument("action", choices=["launch", "status", "destroy"], help="Action to perform (launch, status, destroy)")
     parser.add_argument("--simulation", action="store_true", help="Enable simulation mode")
-    parser.add_argument("--scenario", choices=["scenario-1", "scenario-2"], default="scenario-1", help="Scenario selection")
+    parser.add_argument("--scenario", choices=scenarios, default=scenarios[0], help="Scenario selection")
     return parser.parse_args()
 
+
 def main_function(cloud_provider, action, simulation, scenario):
-    # Call the main function from the imported module and pass the options
     main.main(cloud_provider, action, simulation, scenario)
 
+
 if __name__ == "__main__":
-    display_banner()
     args = parse_arguments()
-
-    # Convert argparse Namespace to dictionary
     options = vars(args)
-
-    # Call the main function with options
     main_function(**options)

core/main.py

@@ -1,98 +1,60 @@
-import os
-import pyfiglet
-import time
-import subprocess
-import json
-from tqdm import tqdm
-from time import sleep
+#! /usr/bin/env python
+# -*- coding: utf-8 -*-
+"""Module providing a class for encapsulating COBRA scenarios."""
 from termcolor import colored
-from .report import gen_report
-from .report import gen_report_2
-from scenarios.scenario_1.scenario_1 import scenario_1_execute
-from scenarios.scenario_2.scenario_2 import scenario_2_execute
-from scenarios.scenario_2.scenario_2 import scenario_2_destroy
 
-def loading_animation():
-    chars = "/—\\|"
-    for _ in range(10):
-        for char in chars:
-            print(f"\rLoading {char}", end="", flush=True)
-            time.sleep(0.1)
+from core.scenario import Scenario
+from helpers.main import print_ascii_art, get_scenarios_config
 
-def select_cloud_provider():
-    print(colored("Select Cloud Provider:", color="yellow"))
-    print(colored("1. AWS", color="green")) 
-    print(colored("2. Azure", color="green"))
-    print(colored("3. GCP", color="green"))
-    while True:
-        try:
-            choice = int(input(colored("Enter your choice (1/2/3): ", color="yellow")))
-            if choice not in [1, 2, 3]:
-                raise ValueError(colored("Invalid choice. Please enter 1, 2, or 3.", color="red"))
-            return choice
-        except ValueError as e:
-            print(e)
 
 def select_attack_scenario(cloud_provider):
-    print(colored("Select Attack Scenario of %s:", color="yellow") % cloud_provider)
-    print(colored("1. Exploit Vulnerable Application, EC2 takeover, Credential Exfiltration & Anomalous Compute Provisioning", color="green"))
-    print(colored("2. Rest API exploit - command injection, credential exfiltration from backend lambda and privilige escalation, rogue identity creation & persistence", color="green"))
+    """Get attack scenario config."""
+    scenarios_config = get_scenarios_config()
+    keys = list(scenarios_config.keys())
+    keys.sort()
+    print(colored('Select Attack Scenario of %s:', color='yellow') % cloud_provider)
+    choices = []
+    for key in keys:
+        index = int(key[-1:])
+        choices.append(index)
+        print(colored('{}. {}: {}'.format(
+            index, scenarios_config[key]['title'], scenarios_config[key]['description']),
+            color='green'))
     while True:
         try:
-            choice = int(input(colored("Enter your choice: ", color="yellow")))
-            if choice not in [1, 2]:
-                raise ValueError(colored("Invalid choice. Please enter 1 or 2.", color="red"))
+            choice = int(input(colored('Enter your choice: ', color='yellow')))
+            if choice not in choices:
+                raise ValueError(colored('Invalid choice.', color='red'))
             return choice
         except ValueError as e:
             print(e)
 
-def get_credentials():
-    while True:
-        try:
-            access_key = input(colored("Enter Access Key: ", color="yellow"))
-            if not access_key:
-                raise ValueError(colored("Access Key cannot be empty.", color="red"))
-            secret_key = input(colored("Enter Secret Key: ", color="yellow"))
-            if not secret_key:
-                raise ValueError(colored("Secret Key cannot be empty.", color="red"))
-            return access_key, secret_key
-        except ValueError as e:
-            print(e)
-
-def execute_scenario(x):
-    try:
-        # Call the scenario function from the imported module
-        if x == 1:
-            scenario_1_execute()
-        elif x == 2:
-            scenario_2_execute()
-        else: 
-            print("Invalid Scenario Selected")
-        print(colored("Scenario executed successfully!", color="green"))
-    except Exception as e:
-        print(colored("Error executing scenario:", color="red"), str(e))
 
 def main(cloud_provider, action, simulation, scenario):
-    if cloud_provider == 'aws':
-        if action == 'launch':
-            if simulation is True:
-                scenario_choice = select_attack_scenario(cloud_provider)
-                if scenario_choice == 1:
-                    # Pass the selected scenario module to execute
-                    execute_scenario(1)
-                elif scenario_choice == 2:
-                    execute_scenario(2)
-                    #print(colored("Scenario coming soon!", color="yellow"))
-        elif action == 'status' and scenario == "scenario-1":
-            subprocess.call("cd ./scenarios/scenario_1/infra/ && pulumi stack ls", shell=True)
-        elif action == 'status' and scenario == "scenario-2":
-            subprocess.call("cd ./scenarios/scenario_2/infra/ && pulumi stack ls", shell=True)
-        elif action == 'destroy' and scenario == "scenario-1":
-            subprocess.call("cd ./scenarios/scenario_1/infra && pulumi destroy", shell=True)
-        elif action == 'destroy' and scenario == "scenario-2":
-            scenario_2_destroy()    
-        else:
-            print('No options provided. --help to know more')
+    """Instantiate and run an attack scenario."""
+    tool_name = 'C O B R A'
+    print_ascii_art(tool_name)
+    scenario_choice = select_attack_scenario(cloud_provider)
+    scenario = Scenario(scenario_choice)
+    if action == 'launch':
+        if simulation:
+            # TODO: what to do with cloud provider?
+            # scenario.setup()
+            scenario.attack()
+            # scenario.destroy()
+            # TODO: ^^^ do we really want to destroy the infra immediately?
+            #   there's a separate command for destroy
+            scenario.generate_report()  # TODO: not implemented
+    elif action == 'status':
+        # TODO
+        # subprocess.call('cd ./scenarios/scenario_2/infra/ && pulumi stack ls', shell=True)
+        pass
+    elif action == 'destroy':
+        scenario.destroy()
+        pass
+    else:
+        print('No options provided. --help to know more')
+
 
-if __name__ == "__main__":
+if __name__ == '__main__':
     main()