archive
/
PurpleDome
mirror of https://github.com/avast/PurpleDome
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

All auto tests work. Integration tests not yet run !

Browse Source
pull/14/head
Thorsten Sick 3 years ago
parent 32af5cf956
commit 724586c60c
4 changed files with 360 additions and 359 deletions
Show Stats Download Patch File Download Diff File

4
app/config_verifier.py
Unescape Escape View File

@ -4,7 +4,7 @@
from pydantic.dataclasses import dataclass from pydantic.dataclasses import dataclass
from pydantic import conlist from pydantic import conlist
from typing import Literal, Optional, TypedDict, Union from typing import Optional
from enum import Enum from enum import Enum
@ -140,4 +140,4 @@ class MainConfig:
return False return False
# TODO: Check for name duplication # TODO: Check for name duplication

492
tests/test_config.py
Unescape Escape View File

@ -23,316 +23,316 @@ class TestMachineConfig(unittest.TestCase):
def test_basic_init(self): def test_basic_init(self):
""" The init is basic and working """ """ The init is basic and working """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1"})) "vm_name": "target1"}))
self.assertEqual(mc.raw_config["root"], "systems/attacker1") self.assertEqual(mc.raw_config["root"], "systems/attacker1")
self.assertEqual(mc.raw_config.vm_controller.vm_type, "vagrant") self.assertEqual(mc.raw_config.vm_controller.vm_type, "vagrant")
def test_use_existing_machine_is_true(self): def test_use_existing_machine_is_true(self):
""" Testing use_existing:machine setting """ """ Testing use_existing:machine setting """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": True})) "use_existing_machine": True}))
self.assertEqual(mc.use_existing_machine(), True) self.assertEqual(mc.use_existing_machine(), True)
def test_use_existing_machine_is_false(self): def test_use_existing_machine_is_false(self):
""" Testing use_existing:machine setting """ """ Testing use_existing:machine setting """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.use_existing_machine(), False) self.assertEqual(mc.use_existing_machine(), False)
def test_use_existing_machine_is_default(self): def test_use_existing_machine_is_default(self):
""" Testing use_existing:machine setting """ """ Testing use_existing:machine setting """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1"})) "vm_name": "target1"}))
self.assertEqual(mc.use_existing_machine(), False) self.assertEqual(mc.use_existing_machine(), False)
def test_windows_is_valid_os(self): def test_windows_is_valid_os(self):
""" Testing if windows is valid os """ """ Testing if windows is valid os """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "windows", "os": "windows",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1"})) "vm_name": "target1"}))
self.assertEqual(mc.os(), "windows") self.assertEqual(mc.os(), "windows")
def test_windows_is_valid_os_casefix(self): def test_windows_is_valid_os_casefix(self):
""" Testing if windows is valid os - using lowercase fix""" """ Testing if windows is valid os - using lowercase fix"""
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "WINDOWS", "os": "WINDOWS",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1"})) "vm_name": "target1"}))
self.assertEqual(mc.os(), "windows") self.assertEqual(mc.os(), "windows")
def test_linux_is_valid_os(self): def test_linux_is_valid_os(self):
""" Testing if windows is valid os """ """ Testing if windows is valid os """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1"})) "vm_name": "target1"}))
self.assertEqual(mc.os(), "linux") self.assertEqual(mc.os(), "linux")
def test_vagrant_is_valid_vmcontroller(self): def test_vagrant_is_valid_vmcontroller(self):
""" Testing if vagrant is valid vmcontroller """ """ Testing if vagrant is valid vmcontroller """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1"})) "vm_name": "target1"}))
self.assertEqual(mc.vmcontroller(), "vagrant") self.assertEqual(mc.vmcontroller(), "vagrant")
def test_vagrant_is_valid_vmcontroller_casefix(self): def test_vagrant_is_valid_vmcontroller_casefix(self):
""" Testing if vagrant is valid vmcontroller case fixxed""" """ Testing if vagrant is valid vmcontroller case fixxed"""
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "VAGRANT", "vm_type": "VAGRANT",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1"})) "vm_name": "target1"}))
self.assertEqual(mc.vmcontroller(), "vagrant") self.assertEqual(mc.vmcontroller(), "vagrant")
def test_vagrant_is_valid_vmip(self): def test_vagrant_is_valid_vmip(self):
""" Testing if vagrant is valid ip/url """ """ Testing if vagrant is valid ip/url """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"ip": "kali", "ip": "kali",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1"})) "vm_name": "target1"}))
self.assertEqual(mc.vm_ip(), "kali") self.assertEqual(mc.vm_ip(), "kali")
def test_missing_vmip(self): def test_missing_vmip(self):
""" Testing if missing vm ip is handled""" """ Testing if missing vm ip is handled"""
vm_name = "target1" vm_name = "target1"
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": vm_name})) "vm_name": vm_name}))
self.assertEqual(mc.vm_ip(), vm_name) self.assertEqual(mc.vm_ip(), vm_name)
def test_machinepath(self): def test_machinepath(self):
""" Testing machinepath setting """ """ Testing machinepath setting """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False, "use_existing_machine": False,
"machinepath": "foo"})) "machinepath": "foo"}))
self.assertEqual(mc.machinepath(), "foo") self.assertEqual(mc.machinepath(), "foo")
def test_machinepath_fallback(self): def test_machinepath_fallback(self):
""" Testing machinepath setting fallback to vmname""" """ Testing machinepath setting fallback to vmname"""
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.machinepath(), "target1") self.assertEqual(mc.machinepath(), "target1")
def test_paw(self): def test_paw(self):
""" Testing for caldera paw """ """ Testing for caldera paw """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"paw": "Bar", "paw": "Bar",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.caldera_paw(), "Bar") self.assertEqual(mc.caldera_paw(), "Bar")
def test_paw_fallback(self): def test_paw_fallback(self):
""" Testing for caldera paw fallback """ """ Testing for caldera paw fallback """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.caldera_paw(), None) self.assertEqual(mc.caldera_paw(), None)
def test_group(self): def test_group(self):
""" Testing for caldera group """ """ Testing for caldera group """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"group": "Bar", "group": "Bar",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.caldera_group(), "Bar") self.assertEqual(mc.caldera_group(), "Bar")
def test_group_fallback(self): def test_group_fallback(self):
""" Testing for caldera group fallback """ """ Testing for caldera group fallback """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.caldera_group(), None) self.assertEqual(mc.caldera_group(), None)
def test_ssh_keyfile(self): def test_ssh_keyfile(self):
""" Testing keyfile config """ """ Testing keyfile config """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"ssh_keyfile": "Bar", "ssh_keyfile": "Bar",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.ssh_keyfile(), "Bar") self.assertEqual(mc.ssh_keyfile(), "Bar")
def test_ssh_keyfile_default(self): def test_ssh_keyfile_default(self):
""" Testing keyfile config default """ """ Testing keyfile config default """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.ssh_keyfile(), None) self.assertEqual(mc.ssh_keyfile(), None)
def test_ssh_user(self): def test_ssh_user(self):
""" Testing ssh user config """ """ Testing ssh user config """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"ssh_user": "Bob", "ssh_user": "Bob",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.ssh_user(), "Bob") self.assertEqual(mc.ssh_user(), "Bob")
def test_ssh_user_default(self): def test_ssh_user_default(self):
""" Testing ssh user default config """ """ Testing ssh user default config """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.ssh_user(), "vagrant") self.assertEqual(mc.ssh_user(), "vagrant")
def test_ssh_password(self): def test_ssh_password(self):
""" Testing ssh password config """ """ Testing ssh password config """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"ssh_user": "Bob", "ssh_user": "Bob",
"ssh_password": "Ross", "ssh_password": "Ross",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.ssh_password(), "Ross") self.assertEqual(mc.ssh_password(), "Ross")
def test_ssh_password_default(self): def test_ssh_password_default(self):
""" Testing ssh password default config """ """ Testing ssh password default config """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertIsNone(mc.ssh_password()) self.assertIsNone(mc.ssh_password())
def test_halt_needs_force_default(self): def test_halt_needs_force_default(self):
""" Testing 'halt needs force' default config """ """ Testing 'halt needs force' default config """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.halt_needs_force(), False) self.assertEqual(mc.halt_needs_force(), False)
def test_halt_needs_force(self): def test_halt_needs_force(self):
""" Testing 'halt needs force' config """ """ Testing 'halt needs force' config """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"halt_needs_force": True, "halt_needs_force": True,
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.halt_needs_force(), True) self.assertEqual(mc.halt_needs_force(), True)
def test_vagrantfilepath(self): def test_vagrantfilepath(self):
""" Testing vagrantfilepath config """ """ Testing vagrantfilepath config """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"halt_needs_force": True, "halt_needs_force": True,
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.vagrantfilepath(), "systems") self.assertEqual(mc.vagrantfilepath(), "systems")
def test_vagrantfilepath_missing(self): def test_vagrantfilepath_missing(self):
@ -340,111 +340,111 @@ class TestMachineConfig(unittest.TestCase):
with self.assertRaises(ConfigurationError): with self.assertRaises(ConfigurationError):
mc = MachineConfig(DotMap(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"halt_needs_force": True, "halt_needs_force": True,
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False}))) "use_existing_machine": False})))
mc.vagrantfilepath() mc.vagrantfilepath()
def test_sensors_empty(self): def test_sensors_empty(self):
""" Testing empty sensor config """ """ Testing empty sensor config """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"halt_needs_force": True, "halt_needs_force": True,
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.sensors(), []) self.assertEqual(mc.sensors(), [])
def test_sensors_set(self): def test_sensors_set(self):
""" Testing empty sensor config """ """ Testing empty sensor config """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"halt_needs_force": True, "halt_needs_force": True,
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False, "use_existing_machine": False,
"sensors": ["linux_idp", "test_sensor"]})) "sensors": ["linux_idp", "test_sensor"]}))
self.assertEqual(mc.sensors(), ["linux_idp", "test_sensor"]) self.assertEqual(mc.sensors(), ["linux_idp", "test_sensor"])
def test_vulnerabilities_empty(self): def test_vulnerabilities_empty(self):
""" Testing empty vulnerabilities config """ """ Testing empty vulnerabilities config """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"halt_needs_force": True, "halt_needs_force": True,
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False})) "use_existing_machine": False}))
self.assertEqual(mc.vulnerabilities(), []) self.assertEqual(mc.vulnerabilities(), [])
def test_vulnerabilities_set(self): def test_vulnerabilities_set(self):
""" Testing empty vulnerabilities config """ """ Testing empty vulnerabilities config """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"halt_needs_force": True, "halt_needs_force": True,
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False, "use_existing_machine": False,
"vulnerabilities": ["PEBKAC", "USER"]})) "vulnerabilities": ["PEBKAC", "USER"]}))
self.assertEqual(mc.vulnerabilities(), ["PEBKAC", "USER"]) self.assertEqual(mc.vulnerabilities(), ["PEBKAC", "USER"])
def test_active_not_set(self): def test_active_not_set(self):
""" machine active not set """ """ machine active not set """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"halt_needs_force": True, "halt_needs_force": True,
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False, "use_existing_machine": False,
"sensors": ["linux_idp", "test_sensor"]})) "sensors": ["linux_idp", "test_sensor"]}))
self.assertEqual(mc.is_active(), True) self.assertEqual(mc.is_active(), True)
def test_active_is_false(self): def test_active_is_false(self):
""" machine active is set to false """ """ machine active is set to false """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"halt_needs_force": True, "halt_needs_force": True,
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False, "use_existing_machine": False,
"active": False, "active": False,
"sensors": ["linux_idp", "test_sensor"]})) "sensors": ["linux_idp", "test_sensor"]}))
self.assertEqual(mc.is_active(), False) self.assertEqual(mc.is_active(), False)
def test_active_is_true(self): def test_active_is_true(self):
""" machine active is set to true """ """ machine active is set to true """
mc = MachineConfig(DotMap({"root": "systems/attacker1", mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"halt_needs_force": True, "halt_needs_force": True,
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
}, },
"vm_name": "target1", "vm_name": "target1",
"use_existing_machine": False, "use_existing_machine": False,
"active": True, "active": True,
"sensors": ["linux_idp", "test_sensor"]})) "sensors": ["linux_idp", "test_sensor"]}))
self.assertEqual(mc.is_active(), True) self.assertEqual(mc.is_active(), True)

220
tests/test_machinecontrol.py
Unescape Escape View File

@ -17,136 +17,136 @@ class TestMachineControl(unittest.TestCase):
def test_get_os_linux_machine(self): def test_get_os_linux_machine(self):
m = Machine(DotMap({"root": "systems/attacker1", m = Machine(DotMap({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"vm_type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target3"}), self.attack_logger)
self.assertEqual(m.get_os(), "linux")
def test_get_os_linux_machine_with_config_class(self):
mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target3"})) "vm_name": "target3"}), self.attack_logger)
self.assertEqual(m.get_os(), "linux")
def test_get_os_linux_machine_with_config_class(self):
mc = MachineConfig(DotMap({"root": "systems/attacker1",
"os": "linux",
"vm_controller": {
"vm_type": "vagrant",
"vagrantfilepath": "systems",
},
"vm_name": "target3"}))
m = Machine(mc, self.attack_logger) m = Machine(mc, self.attack_logger)
self.assertEqual(m.get_os(), "linux") self.assertEqual(m.get_os(), "linux")
def test_get_paw_good(self): def test_get_paw_good(self):
m = Machine(DotMap({"root": "systems/attacker1", m = Machine(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"paw": "testme", "paw": "testme",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target3"}), self.attack_logger) "vm_name": "target3"}), self.attack_logger)
self.assertEqual(m.get_paw(), "testme") self.assertEqual(m.get_paw(), "testme")
def test_get_paw_missing(self): def test_get_paw_missing(self):
m = Machine(DotMap({"root": "systems/attacker1", m = Machine(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target3" "vm_name": "target3"
}), self.attack_logger) }), self.attack_logger)
self.assertEqual(m.get_paw(), None) self.assertEqual(m.get_paw(), None)
def test_get_group_good(self): def test_get_group_good(self):
m = Machine(DotMap({"root": "systems/attacker1", m = Machine(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"group": "testme", "group": "testme",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target3"}), self.attack_logger) "vm_name": "target3"}), self.attack_logger)
self.assertEqual(m.get_group(), "testme") self.assertEqual(m.get_group(), "testme")
def test_get_group_missing(self): def test_get_group_missing(self):
m = Machine(DotMap({"root": "systems/attacker1", m = Machine(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target3" "vm_name": "target3"
}), self.attack_logger) }), self.attack_logger)
self.assertEqual(m.get_group(), None) self.assertEqual(m.get_group(), None)
def test_vagrantfilepath_missing(self): def test_vagrantfilepath_missing(self):
with self.assertRaises(ConfigurationError): with self.assertRaises(ConfigurationError):
Machine(DotMap({"root": "systems/attacker1", Machine(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
}, },
"vm_name": "target3" "vm_name": "target3"
}), self.attack_logger) }), self.attack_logger)
def test_vagrantfile_missing(self): def test_vagrantfile_missing(self):
with self.assertRaises(ConfigurationError): with self.assertRaises(ConfigurationError):
Machine(DotMap({"root": "systems/attacker1", Machine(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "non_existing", "vagrantfilepath": "non_existing",
}, },
"vm_name": "target3" "vm_name": "target3"
}), self.attack_logger) }), self.attack_logger)
def test_vagrantfile_existing(self): def test_vagrantfile_existing(self):
m = Machine(DotMap({"root": "systems/attacker1", m = Machine(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target3" "vm_name": "target3"
}), self.attack_logger) }), self.attack_logger)
self.assertIsNotNone(m) self.assertIsNotNone(m)
# test: auto generated, dir missing # test: auto generated, dir missing
def test_auto_generated_machinepath_with_path_missing(self): def test_auto_generated_machinepath_with_path_missing(self):
with self.assertRaises(ConfigurationError): with self.assertRaises(ConfigurationError):
Machine(DotMap({"root": "systems/attacker1", Machine(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "missing" "vm_name": "missing"
}), self.attack_logger) }), self.attack_logger)
# test manual config, dir missing # test manual config, dir missing
def test_configured_machinepath_with_path_missing(self): def test_configured_machinepath_with_path_missing(self):
with self.assertRaises(ConfigurationError): with self.assertRaises(ConfigurationError):
Machine(DotMap({"root": "systems/attacker1", Machine(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target3", "vm_name": "target3",
"machinepath": "missing" "machinepath": "missing"
}), self.attack_logger) }), self.attack_logger)
# test auto generated, dir there (external/internal dirs must work !) # test auto generated, dir there (external/internal dirs must work !)
def test_auto_generated_machinepath_with_good_config(self): def test_auto_generated_machinepath_with_good_config(self):
m = Machine(DotMap({"root": "systems/attacker1", m = Machine(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target3" "vm_name": "target3"
}), self.attack_logger) }), self.attack_logger)
vagrantfilepath = os.path.abspath("systems") vagrantfilepath = os.path.abspath("systems")
ext = os.path.join(vagrantfilepath, "target3") ext = os.path.join(vagrantfilepath, "target3")
internal = os.path.join("/vagrant/", "target3") internal = os.path.join("/vagrant/", "target3")
@ -157,14 +157,14 @@ class TestMachineControl(unittest.TestCase):
# test: manual config, dir there (external/internal dirs must work !) # test: manual config, dir there (external/internal dirs must work !)
def test_configured_machinepath_with_good_config(self): def test_configured_machinepath_with_good_config(self):
m = Machine(DotMap({"root": "systems/attacker1", m = Machine(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "missing", "vm_name": "missing",
"machinepath": "target3" "machinepath": "target3"
}), self.attack_logger) }), self.attack_logger)
vagrantfilepath = os.path.abspath("systems") vagrantfilepath = os.path.abspath("systems")
ext = os.path.join(vagrantfilepath, "target3") ext = os.path.join(vagrantfilepath, "target3")
internal = os.path.join("/vagrant/", "target3") internal = os.path.join("/vagrant/", "target3")
@ -176,22 +176,22 @@ class TestMachineControl(unittest.TestCase):
def test_configured_vm_controller_missing(self): def test_configured_vm_controller_missing(self):
with self.assertRaises(ConfigurationError): with self.assertRaises(ConfigurationError):
Machine(DotMap({"root": "systems/attacker1", Machine(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_name": "missing", "vm_name": "missing",
"machinepath": "target3" "machinepath": "target3"
}), self.attack_logger) }), self.attack_logger)
# Create caldera start command and verify it # Create caldera start command and verify it
def test_get_linux_caldera_start_cmd(self): def test_get_linux_caldera_start_cmd(self):
m = Machine(DotMap({"root": "systems/attacker1", m = Machine(DotMap({"root": "systems/attacker1",
"os": "linux", "os": "linux",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target3", "vm_name": "target3",
"group": "testgroup", "group": "testgroup",
"paw": "testpaw"}), self.attack_logger) "paw": "testpaw"}), self.attack_logger)
m.set_caldera_server("http://www.test.test") m.set_caldera_server("http://www.test.test")
with patch.object(m.vm_manager, "get_playground", return_value="/vagrant/target3"): with patch.object(m.vm_manager, "get_playground", return_value="/vagrant/target3"):
cmd = m.create_start_caldera_client_cmd() cmd = m.create_start_caldera_client_cmd()
@ -200,15 +200,15 @@ class TestMachineControl(unittest.TestCase):
# Create caldera start command and verify it (windows) # Create caldera start command and verify it (windows)
def test_get_windows_caldera_start_cmd(self): def test_get_windows_caldera_start_cmd(self):
m = Machine(DotMap({"root": "systems/attacker1", m = Machine(DotMap({"root": "systems/attacker1",
"os": "windows", "os": "windows",
"vm_controller": { "vm_controller": {
"vm_type": "vagrant", "vm_type": "vagrant",
"vagrantfilepath": "systems", "vagrantfilepath": "systems",
}, },
"vm_name": "target3", "vm_name": "target3",
"group": "testgroup", "group": "testgroup",
"paw": "testpaw", "paw": "testpaw",
"machinepath": "target2w"}), self.attack_logger) "machinepath": "target2w"}), self.attack_logger)
m.set_caldera_server("www.test.test") m.set_caldera_server("www.test.test")
cmd = m.create_start_caldera_client_cmd() cmd = m.create_start_caldera_client_cmd()
self.maxDiff = None self.maxDiff = None

3
tox.ini
Unescape Escape View File

@ -44,7 +44,8 @@ commands =
# Ignoring: # Ignoring:
# C901 complex code. Reduce complexitiy. But this thing is over-reacting # C901 complex code. Reduce complexitiy. But this thing is over-reacting
# E501: line too long. Please: Still keep it short. But 80 chars is just incredibly short nowadays # E501: line too long. Please: Still keep it short. But 80 chars is just incredibly short nowadays
flake8 --ignore C901,E501 # W601: TODO: has_key in config_verifier.py should be replaced by an iterator
flake8 --ignore C901,E501,W601
# Check if dependencies are vulnerable # Check if dependencies are vulnerable
safety check -r requirements.txt safety check -r requirements.txt
# Check for common vulnerabilities # Check for common vulnerabilities

Write Preview
Loading…
Cancel
Save