File size: 3,596 Bytes
7c89ed7 | 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 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | from .positive_alert_test_case import PositiveAlertTestCase
from .negative_alert_test_case import NegativeAlertTestCase
from .alert_test_suite import AlertTestSuite
class TestAlertSSHAccess(AlertTestSuite):
alert_classname = "AlertSSHAccess"
alert_filename = "ssh_access"
# This event is the default positive event that will cause the
# alert to trigger
default_event = {
"_source": {
"category": "syslog",
"hostname": 'victim1.small.corp.com',
"summary": 'Accepted publickey for alamakota from 11.22.33.44 port 39190 ssh2',
"details": {
"sourceipaddress": "11.22.33.44",
"program": "sshd",
"username": "alamakota"
}
}
}
# This alert is the expected result from running this task
default_alert = {
"category": "access",
"severity": "CRITICAL",
"summary": "SSH login from 11.22.33.44 on victim1.small.corp.com as user alamakota",
"tags": ['ssh'],
"notify_mozdefbot": True
}
test_cases = []
event = AlertTestSuite.create_event(default_event)
test_cases.append(
PositiveAlertTestCase(
description="Positive test case with good event",
events=[event],
expected_alert=default_alert
)
)
event = AlertTestSuite.create_event(default_event)
event['_source']['details']['sourceipaddress'] = "11.22.33.45"
newip_alert = default_alert.copy()
newip_alert['summary'] = "SSH login from 11.22.33.45 on victim1.small.corp.com as user alamakota"
test_cases.append(
PositiveAlertTestCase(
description="Positive test case from a different server",
events=[event],
expected_alert=newip_alert
)
)
event = AlertTestSuite.create_event(default_event)
event['_source']['utctimestamp'] = AlertTestSuite.subtract_from_timestamp_lambda({'minutes': 14})
event['_source']['receivedtimestamp'] = AlertTestSuite.subtract_from_timestamp_lambda({'minutes': 14})
test_cases.append(
PositiveAlertTestCase(
description="Positive test case with an event with somewhat old timestamp",
events=[event],
expected_alert=default_alert
)
)
event = AlertTestSuite.create_event(default_event)
event['_source']['utctimestamp'] = AlertTestSuite.subtract_from_timestamp_lambda({'minutes': 16})
event['_source']['receivedtimestamp'] = AlertTestSuite.subtract_from_timestamp_lambda({'minutes': 16})
test_cases.append(
NegativeAlertTestCase(
description="Negative test case with old timestamp",
events=[event],
)
)
event = AlertTestSuite.create_event(default_event)
event['_source']['category'] = ['bro']
test_cases.append(
NegativeAlertTestCase(
description="Negative test case with bad category",
events=[event],
)
)
event = AlertTestSuite.create_event(default_event)
event['_source']['details']['program'] = 'ssh'
test_cases.append(
NegativeAlertTestCase(
description="Negative test case with bad program",
events=[event],
)
)
event = AlertTestSuite.create_event(default_event)
event['_source']['details']['sourceipaddress'] = '213.212.11.5'
test_cases.append(
NegativeAlertTestCase(
description="Negative test case with the source IP address outside of the watchlist",
events=[event],
)
)
|