archive
/
PurpleDome
mirror of https://github.com/avast/PurpleDome
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
develop
polishing2
presentation_gpn_2022
more_unit_tests
extending_page
main
caldera_4
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '3bac1d862d'
${ noResults }
PurpleDome/tests/test_config.py

670 lines
27 KiB
Python
Raw Normal View History Unescape Escape

Unit tests working now
4 years ago
#!/usr/bin/env python3
# Testing the central configuration loader
import unittest
# import os
from app.config import ExperimentConfig, MachineConfig
from app.exceptions import ConfigurationError
# https://docs.python.org/3/library/unittest.html
class TestMachineConfig(unittest.TestCase):
""" Test machine specific config """
def test_empty_init(self):
""" The init is empty """
with self.assertRaises(ConfigurationError):
MachineConfig(None)
def test_basic_init(self):
""" The init is basic and working """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1"})
self.assertEqual(mc.raw_config["root"], "systems/attacker1")
self.assertEqual(mc.raw_config["vm_controller"]["type"], "vagrant")
def test_missing_vm_name(self):
""" The vm name is missing """
with self.assertRaises(ConfigurationError):
MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
}})
def test_use_existing_machine_is_true(self):
""" Testing use_existing:machine setting """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": True})
self.assertEqual(mc.use_existing_machine(), True)
def test_use_existing_machine_is_false(self):
""" Testing use_existing:machine setting """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.use_existing_machine(), False)
def test_use_existing_machine_is_default(self):
""" Testing use_existing:machine setting """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1"})
self.assertEqual(mc.use_existing_machine(), False)
def test_windows_is_valid_os(self):
""" Testing if windows is valid os """
mc = MachineConfig({"root": "systems/attacker1",
"os": "windows",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1"})
self.assertEqual(mc.os(), "windows")
def test_windows_is_valid_os_casefix(self):
""" Testing if windows is valid os - using lowercase fix"""
mc = MachineConfig({"root": "systems/attacker1",
"os": "WINDOWS",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1"})
self.assertEqual(mc.os(), "windows")
def test_linux_is_valid_os(self):
""" Testing if windows is valid os """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1"})
self.assertEqual(mc.os(), "linux")
def test_missing_os(self):
""" The os is missing """
with self.assertRaises(ConfigurationError):
MachineConfig({"root": "systems/attacker1",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1"})
def test_wrong_os(self):
""" The os is wrong """
with self.assertRaises(ConfigurationError):
MachineConfig({"root": "systems/attacker1",
"os": "BROKEN",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1"})
def test_vagrant_is_valid_vmcontroller(self):
""" Testing if vagrant is valid vmcontroller """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1"})
self.assertEqual(mc.vmcontroller(), "vagrant")
def test_vagrant_is_valid_vmcontroller_casefix(self):
""" Testing if vagrant is valid vmcontroller case fixxed"""
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "VAGRANT",
"vagrantfilepath": "systems",
},
"vm_name": "target1"})
self.assertEqual(mc.vmcontroller(), "vagrant")
def test_invalid_vmcontroller(self):
""" Testing if vagrant is valid vmcontroller case fixxed"""
with self.assertRaises(ConfigurationError):
MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "BROKEN",
"vagrantfilepath": "systems",
},
"vm_name": "target1"})
def test_missing_vmcontroller_2(self):
""" Testing if vagrant is valid vmcontroller case fixxed"""
with self.assertRaises(ConfigurationError):
MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_name": "target1"})
def test_vagrant_is_valid_vmip(self):
""" Testing if vagrant is valid ip/url """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"ip": "kali",
"vagrantfilepath": "systems",
},
"vm_name": "target1"})
self.assertEqual(mc.vm_ip(), "kali")
def test_missing_vmip(self):
""" Testing if missing vm ip is handled"""
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1"})
self.assertEqual(mc.vm_ip(), None)
def test_machinepath(self):
""" Testing machinepath setting """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False,
"machinepath": "foo"})
self.assertEqual(mc.machinepath(), "foo")
def test_machinepath_fallback(self):
""" Testing machinepath setting fallback to vmname"""
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.machinepath(), "target1")
def test_paw(self):
""" Testing for caldera paw """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"paw": "Bar",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.caldera_paw(), "Bar")
def test_paw_fallback(self):
""" Testing for caldera paw fallback """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.caldera_paw(), None)
def test_group(self):
""" Testing for caldera group """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"group": "Bar",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.caldera_group(), "Bar")
def test_group_fallback(self):
""" Testing for caldera group fallback """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.caldera_group(), None)
def test_ssh_keyfile(self):
""" Testing keyfile config """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"ssh_keyfile": "Bar",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.ssh_keyfile(), "Bar")
def test_ssh_keyfile_default(self):
""" Testing keyfile config default """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.ssh_keyfile(), None)
def test_ssh_user(self):
""" Testing ssh user config """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"ssh_user": "Bob",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.ssh_user(), "Bob")
def test_ssh_user_default(self):
""" Testing ssh user default config """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.ssh_user(), "vagrant")
Added the option to supply a ssh password
4 years ago
def test_ssh_password(self):
""" Testing ssh password config """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"ssh_user": "Bob",
"ssh_password": "Ross",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.ssh_password(), "Ross")
def test_ssh_password_default(self):
""" Testing ssh password default config """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertIsNone(mc.ssh_password())
Unit tests working now
4 years ago
def test_halt_needs_force_default(self):
""" Testing 'halt needs force' default config """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.halt_needs_force(), False)
def test_halt_needs_force(self):
""" Testing 'halt needs force' config """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"halt_needs_force": True,
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.halt_needs_force(), True)
def test_vagrantfilepath(self):
""" Testing vagrantfilepath config """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"halt_needs_force": True,
"vm_controller": {
"type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.vagrantfilepath(), "systems")
def test_vagrantfilepath_missing(self):
""" Testing missing vagrantfilepath config """
with self.assertRaises(ConfigurationError):
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"halt_needs_force": True,
"vm_controller": {
"type": "vagrant",
},
"vm_name": "target1",
"use_existing_machine": False})
mc.vagrantfilepath()
def test_sensors_empty(self):
""" Testing empty sensor config """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"halt_needs_force": True,
"vm_controller": {
"type": "vagrant",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.sensors(), [])
def test_sensors_set(self):
""" Testing empty sensor config """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"halt_needs_force": True,
"vm_controller": {
"type": "vagrant",
},
"vm_name": "target1",
"use_existing_machine": False,
"sensors": ["linux_idp", "test_sensor"]})
self.assertEqual(mc.sensors(), ["linux_idp", "test_sensor"])
def test_vulnerabilities_empty(self):
""" Testing empty vulnerabilities config """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"halt_needs_force": True,
"vm_controller": {
"type": "vagrant",
},
"vm_name": "target1",
"use_existing_machine": False})
self.assertEqual(mc.vulnerabilities(), [])
def test_vulnerabilities_set(self):
""" Testing empty vulnerabilities config """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"halt_needs_force": True,
"vm_controller": {
"type": "vagrant",
},
"vm_name": "target1",
"use_existing_machine": False,
"vulnerabilities": ["PEBKAC", "USER"]})
self.assertEqual(mc.vulnerabilities(), ["PEBKAC", "USER"])
def test_active_not_set(self):
""" machine active not set """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"halt_needs_force": True,
"vm_controller": {
"type": "vagrant",
},
"vm_name": "target1",
"use_existing_machine": False,
"sensors": ["linux_idp", "test_sensor"]})
self.assertEqual(mc.is_active(), True)
def test_active_is_false(self):
""" machine active is set to false """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"halt_needs_force": True,
"vm_controller": {
"type": "vagrant",
},
"vm_name": "target1",
"use_existing_machine": False,
"active": False,
"sensors": ["linux_idp", "test_sensor"]})
self.assertEqual(mc.is_active(), False)
def test_active_is_true(self):
""" machine active is set to true """
mc = MachineConfig({"root": "systems/attacker1",
"os": "linux",
"halt_needs_force": True,
"vm_controller": {
"type": "vagrant",
},
"vm_name": "target1",
"use_existing_machine": False,
"active": True,
"sensors": ["linux_idp", "test_sensor"]})
self.assertEqual(mc.is_active(), True)
class TestExperimentConfig(unittest.TestCase):
def test_missing_config_file(self):
""" Missing config file """
with self.assertRaises(FileNotFoundError):
ExperimentConfig("tests/data/missing.yaml")
def test_basic_loading(self):
""" Existing, basic config file, testing the values are loaded properly """
ex = ExperimentConfig("tests/data/basic.yaml")
self.assertEqual(ex.raw_config["caldera"]["apikey"], "ADMIN123")
self.assertEqual(ex.caldera_apikey(), "ADMIN123")
def test_basic_loading_targets_read(self):
""" Targets in config: can be found """
ex = ExperimentConfig("tests/data/basic.yaml")
self.assertEqual(len(ex._targets), 2)
self.assertEqual(ex._targets[0].raw_config["vm_name"], "target1")
self.assertEqual(ex._targets[0].vmname(), "target1")
self.assertEqual(ex.targets()[0].vmname(), "target1")
def test_basic_loading_attacker_read(self):
""" Attackers in config: can be found """
ex = ExperimentConfig("tests/data/basic.yaml")
self.assertEqual(len(ex._targets), 2)
self.assertEqual(ex._attackers[0].raw_config["vm_name"], "attacker")
self.assertEqual(ex._attackers[0].vmname(), "attacker")
self.assertEqual(ex.attackers()[0].vmname(), "attacker")
self.assertEqual(ex.attacker(0).vmname(), "attacker")
def test_missing_kali_config(self):
""" Getting kali config for a specific attack. Attack missing """
ex = ExperimentConfig("tests/data/basic.yaml")
Adjusted unit tests to current code
3 years ago
self.assertEqual(ex.kali_conf("BOOM"), {})
Unit tests working now
4 years ago
def test_working_kali_config(self):
""" Getting kali config for a specific attack """
ex = ExperimentConfig("tests/data/basic.yaml")
data = ex.kali_conf("hydra")
self.assertEqual(data["userfile"], "users.txt")
Caldera now supports jitter and obfuscator from configuration file. Keep in mind: Not all implants support all obfuscators.
4 years ago
def test_missing_caldera_config_obfuscator(self):
""" A config file with no caldera config at all """
ex = ExperimentConfig("tests/data/basic.yaml")
self.assertEqual(ex.get_caldera_obfuscator(), "plain-text")
def test_broken_caldera_config_obfuscator(self):
""" A config file with broken caldera config at all """
ex = ExperimentConfig("tests/data/partial.yaml")
self.assertEqual(ex.get_caldera_obfuscator(), "plain-text")
def test_good_caldera_config_obfuscator(self):
""" A config file with broken caldera config at all """
ex = ExperimentConfig("tests/data/attacks_perfect.yaml")
self.assertEqual(ex.get_caldera_obfuscator(), "foo-bar")
def test_missing_caldera_config_jitter(self):
""" A config file with no caldera config at all """
ex = ExperimentConfig("tests/data/basic.yaml")
self.assertEqual(ex.get_caldera_jitter(), "4/8")
def test_broken_caldera_config_jitter(self):
""" A config file with broken caldera config at all """
ex = ExperimentConfig("tests/data/partial.yaml")
self.assertEqual(ex.get_caldera_jitter(), "4/8")
def test_good_caldera_config_jitter(self):
""" A config file with broken caldera config at all """
ex = ExperimentConfig("tests/data/attacks_perfect.yaml")
self.assertEqual(ex.get_caldera_jitter(), "08/15")
Unit tests working now
4 years ago
def test_nap_time(self):
""" nap time is set """
ex = ExperimentConfig("tests/data/basic.yaml")
self.assertEqual(ex.get_nap_time(), 5)
def test_nap_time_not_set(self):
""" nap time is not set """
ex = ExperimentConfig("tests/data/nap_time_missing.yaml")
self.assertEqual(ex.get_nap_time(), 0)
Fixing bug: Zero kali or caldera attacks crash the script
4 years ago
def test_kali_attacks_missing(self):
""" kali attacks entry fully missing from config """
ex = ExperimentConfig("tests/data/attacks_missing.yaml")
self.assertEqual(ex.get_kali_attacks("linux"), [])
def test_kali_attacks_empty(self):
""" zero entries in kali attacks list """
ex = ExperimentConfig("tests/data/attacks_perfect.yaml")
self.assertEqual(ex.get_kali_attacks("missing"), [])
def test_kali_attacks_one(self):
""" One entry in kali attacks list """
ex = ExperimentConfig("tests/data/attacks_perfect.yaml")
self.assertEqual(ex.get_kali_attacks("linux"), ["hydra"])
def test_kali_attacks_many(self):
""" Many entries in kali attacks list """
ex = ExperimentConfig("tests/data/attacks_perfect.yaml")
self.assertEqual(ex.get_kali_attacks("windows"), ["hydra", "medusa", "skylla"])
def test_caldera_attacks_missing(self):
""" caldera attacks entry fully missing from config """
ex = ExperimentConfig("tests/data/attacks_missing.yaml")
self.assertEqual(ex.get_caldera_attacks("linux"), [])
Added bug with partial attack configs (kali or caldera) in yaml file
4 years ago
def test_kali_attacks_half(self):
""" kali attacks entry partially missing from config """
ex = ExperimentConfig("tests/data/attacks_half.yaml")
self.assertEqual(ex.get_kali_attacks("linux"), ["hydra"])
self.assertEqual(ex.get_kali_attacks("windows"), [])
def test_caldera_attacks_half(self):
""" caldera attacks entry partially missing from config """
ex = ExperimentConfig("tests/data/attacks_half.yaml")
self.assertEqual(ex.get_caldera_attacks("linux"), ["bd527b63-9f9e-46e0-9816-b8434d2b8989"])
self.assertEqual(ex.get_caldera_attacks("windows"), [])
Fixing bug: Zero kali or caldera attacks crash the script
4 years ago
def test_caldera_attacks_empty(self):
""" zero entries in caldera attacks list """
ex = ExperimentConfig("tests/data/attacks_perfect.yaml")
self.assertEqual(ex.get_caldera_attacks("missing"), [])
def test_caldera_attacks_one(self):
""" One entry in caldera attacks list """
ex = ExperimentConfig("tests/data/attacks_perfect.yaml")
self.assertEqual(ex.get_caldera_attacks("linux"), ["bd527b63-9f9e-46e0-9816-b8434d2b8989"])
def test_caldera_attacks_many(self):
""" Many entries in caldera attacks list """
ex = ExperimentConfig("tests/data/attacks_perfect.yaml")
self.assertEqual(ex.get_caldera_attacks("windows"), ["bd527b63-9f9e-46e0-9816-b8434d2b8989", "foo", "bar"])
Unit tests working now
4 years ago
if __name__ == '__main__':
unittest.main()