File size: 4,404 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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
# Copyright (c) 2017 Mozilla Corporation
from .positive_alert_test_case import PositiveAlertTestCase
from .negative_alert_test_case import NegativeAlertTestCase

from .alert_test_suite import AlertTestSuite


class TestAlertProxyDropExecutable(AlertTestSuite):
    alert_filename = "proxy_drop_executable"
    # This event is the default positive event that will cause the
    # alert to trigger
    default_event = {
        "_source": {
            "category": "proxy",
            "details": {
                "sourceipaddress": "1.2.3.4",
                "destination": "http://evil.com/evil.exe",
                "proxyaction": "TCP_DENIED",
            },
        },
    }

    # This event is an alternate destination that we'd want to aggregate
    default_event2 = AlertTestSuite.copy(default_event)
    default_event2["_source"]["details"]["destination"] = "http://evil.com/evil.sh"

    # This event is the default negative event that will not cause the
    # alert to trigger
    default_negative_event = AlertTestSuite.copy(default_event)
    default_negative_event["_source"]["details"][
        "destination"
    ] = "http://foo.mozilla.com/index.html"

    # This alert is the expected result from running this task
    default_alert = {
        "category": "squid",
        "tags": ["squid", "proxy"],
        "severity": "WARNING",
        "summary": "Suspicious Proxy DROP event(s) detected from 1.2.3.4 to the following executable file destination(s): http://evil.com/evil.exe",
    }

    # This alert is the expected result from this task against multiple matching events
    default_alert_aggregated = AlertTestSuite.copy(default_alert)
    default_alert_aggregated[
        "summary"
    ] = "Suspicious Proxy DROP event(s) detected from 1.2.3.4 to the following executable file destination(s): http://evil.com/evil.exe,http://evil.com/evil.sh"

    test_cases = []

    test_cases.append(
        PositiveAlertTestCase(
            description="Positive test with default events and default alert expected",
            events=AlertTestSuite.create_events(default_event, 1),
            expected_alert=default_alert,
        )
    )

    test_cases.append(
        PositiveAlertTestCase(
            description="Positive test with default events and default alert expected - dedup",
            events=AlertTestSuite.create_events(default_event, 2),
            expected_alert=default_alert,
        )
    )

    events1 = AlertTestSuite.create_events(default_event, 1)
    events2 = AlertTestSuite.create_events(default_event2, 1)
    test_cases.append(
        PositiveAlertTestCase(
            description="Positive test with default events and default alert expected - different dests",
            events=events1 + events2,
            expected_alert=default_alert_aggregated,
        )
    )

    test_cases.append(
        NegativeAlertTestCase(
            description="Negative test with default negative event",
            events=AlertTestSuite.create_events(default_negative_event, 1),
        )
    )

    events = AlertTestSuite.create_events(default_event, 10)
    for event in events:
        event["_source"]["category"] = "bad"
    test_cases.append(
        NegativeAlertTestCase(
            description="Negative test case with events with incorrect category",
            events=events,
        )
    )

    events = AlertTestSuite.create_events(default_event, 10)
    for event in events:
        event["_source"][
            "utctimestamp"
        ] = AlertTestSuite.subtract_from_timestamp_lambda({"minutes": 241})
        event["_source"][
            "receivedtimestamp"
        ] = AlertTestSuite.subtract_from_timestamp_lambda({"minutes": 241})
    test_cases.append(
        NegativeAlertTestCase(
            description="Negative test case with old timestamp", events=events
        )
    )

    events = AlertTestSuite.create_events(default_event, 10)
    for event in events:
        event["_source"]["details"]["destination"] = "http://evil.com/evil.pdf"
    test_cases.append(
        NegativeAlertTestCase(
            description="Negative test case with events with non blacklisted extension",
            events=events,
        )
    )