<jupyter_start><jupyter_text>Looking beyond the details: understand attack 'storylines' with MITRE tags  Introduction Today's compute environments are typically well monitored by multiple different network- and host-based sensors - producing a large amount of events and alerts even after careful tuning.In case of a complex attack chain, where an attacker is able to break into an internal system and move to other systems, gain privileges and get access to data, different alerts will be created during the attacker's activities.While the detailed monitoring is great to catch and mitigate very specific attack steps, it can be hard to get the bigger picture of the attack operation and understand the overall goals (and results) of the attack.In the following, we explore one possible way to abstract from the details to a higher-level understanding of an attack chain to expose the 'storyline' of the attack with the help of MITRE tags - complementary to the lower-level details of alerts.The [MITRE ATT&CK®](https://attack.mitre.org/) framework defines a "globally-accessible knowledge base of adversary tactics and techniques based on real-world observations", i.e., describes the different kinds of possible attack steps in a systematic way, independent from any particular tool. For each of the documented techniques/sub-techniques the MITRE ATT&CK includes textual context, examples, ways of mitigations and detections and references. It is structured into a set of 14 high-level tactics, like reconnaissance, initial access, privileged escalation etc, each of which is comprised of many techniques and sub-techniques in a hierarchical fashion. One concrete example for illustration is the [Command and Scripting Interpreter: PowerShell](https://attack.mitre.org/techniques/T1059/001/) with the MITRE tag T1059.001. It describes the potential abuse of the PowerShell by adversaries for execution of malicious tasks. The tag T1059.001 indicates that this is the sub-technique 1 of the technique [T1059: Command and Scripting Interpreter ](https://attack.mitre.org/techniques/T1059/) that in turn is part of the high-level tactic [Execution](https://attack.mitre.org/tactics/TA0002/).Particularly important is that this effort is independent of any specific tool and thereby can be used to understand attack steps across multiple tools. Support for tagging the tool-specific alerts with MITRE tags are in many cases under way or already available.Intuition suggests that a more complex attack chain will have some typical sequence of phases at a higher level, proceeding maybe in progression like discovery → initial access → privilege escalation → data exfiltration → persistence (although this in general will not be a strict sequence of phases, as an attacker might skip or repeat phases, e.g., after further lateral movements). To test and illustrate this approach we have created a small test environment. In this setup, we use a known, scripted attack scenario while observing the environment with multiple sensors and collect the alerts generated by the sensors. Instead of looking at the details of these alerts, we here go in the opposite direction of abstracting from the low-level, detailed alerts to look only at the timeline of the MITRE tags associated with the alerts.  Experimental setup  The small environment that we use for testing is shown in the above figure: the blue parts are the 'productive' components comprised of a web application `juice-shop`, a MySQL server and MS Windows server. The web application we use is the intentionally vulnerable web shop created by the [OWASP](https://owasp.org/) team called[Juice Shop](https://owasp.org/www-project-juice-shop) for testing and learning. We use it on a Linux VM and on the Windows VM as an easy way for us to create an attack scenario targetting some of the known vulnerabilities of the Juice-Shop. In addition, there are a small number of VMs that act as simulated users of the webshop in order to create 'normal' background traffic.The monitoring components are shown in green: we monitor the Linux VMs with SysFlow, the network is monitored by Snort and for the Windows server we have a setup that feeds Sysmon events through Winlogbeat into an instance of [HELK](https://github. com/Cyb3rWard0g/HELK). In this article, we only look at alerts (rather then the base telemetry events produced by SysFlow/Sysmon) that are generated by a TTP ruleset on the SysFlow side and the set of [Sigma](https://github.com/SigmaHQ/sigma) rules on the Sysmon side. The alerts based on the Sysmon events are in this setting created by [ElastAlert](https://github.com/Yelp/elastalert) that is part of the HELK package and is configured to use the set of Sigma rules.With regards to the current availability of MITRE tags: - the TTP ruleset used by SysFlow already contains MITRE tags, and the team is continually extending the existing list- also Sigma rules already are enriched with MITRE tags that we can use in ElastAlert- Snort rules currently have no MITRE tags attached currently - in this case we manually added MITRE tag information to selected rules.Our setup also contains a transformation of the alerts into the [Elastic Common Schema (ECS)](https://www. elastic.co/what-is/ecs/) format for normalisation as part of the pipeline before they are collected in two instances of ElasticSearch.For SysFlow this capability to convert events and alerts to the ECS format and storage into ElasticSearch was added in version 0.3.0 - you can find details about it in the [blog](https://sysflow.io/2021/08/20/elk-integration.html) together with a detailed [demonstration](https://github.com/sysflow-telemetry/sf-deployments/tree/dev/integrations/elk). The figure below shows the attack scenario that we created as a scripted way to run a complex attack scenario in a reproducible way: it contains steps from reconnaisance, like nmap scans, uses some of the known intentional vulnerabilities of the OWASP Juice-Shop to break into the web shop, attack from this foothold the database server and exfiltrate data from it. As a next step, the attacker moves laterally to the Windows server where the attack succeeds to install an additional user and a persistent backdoor with OpenSSH.   Create MITRE tactics timeline from ECS alert data  Define mapping MITRE tag → MITRE tactic The alerts received by our sensors contain MITRE tags like T1059.001, but from this tag alone we do not know the MITRE tactic it belongs to as there is no strict hierarchy as MITRE techniques can belong to multiple tactics.Luckily there is an API for the MITRE ATT&CK data that we can use to resolve the tag to the tactic(s) it belongs to.To achieve this we first download the information about all techniques via the API and then build a mapping `tag2tactics` for each tag to the tactics it belongs to.<jupyter_code># --- imports as required
import json, os, collections, datetime
import pandas as pd
from taxii2client.v20 import Server, Collection
from stix2 import TAXIICollectionSource, Filter
import plotly.graph_objects as go
import plotly as pl
import plotly.io as pio
pio.renderers.default = 'iframe'
# ----- download the ATT&CK data via its API
def download_MITRE_data():
    # see https://www.mitre.org/capabilities/cybersecurity/overview/cybersecurity-blog/attck%E2%84%A2-content-available-in-stix%E2%84%A2-20-via

    # Instantiate server and get API Root
    server = Server("https://cti-taxii.mitre.org/taxii/")
    api_root = server.api_roots[0]

    # Print name and ID of all ATT&CK technology-domains available as collections
    for collection in api_root.collections:
        print(collection.title + ": " + collection.id)

    # Establish TAXII2 Collection instance for Enterprise ATT&CK collection
    collection = Collection("https://cti-taxii.mitre.org/stix/collections/95ecc380-afe9-11e4-9b6c-751b66dd541e/")

    # Supply the collection to TAXIICollection
    tc_source = TAXIICollectionSource(collection)

    # Fetch information about the techniques
    techniques = tc_source.query([Filter("type", "=", "attack-pattern")])
    return techniques

# ----- create a Python dict 'tag2tactics' containing the mapping
def get_mitre_tag(technique):
    external_refs = technique.get('external_references')
    mitre_tags = list(map(lambda x: x.get('external_id'), 
                   filter(lambda x: x.get('source_name') == 'mitre-attack', external_refs)
                  ))
    if len(mitre_tags) == 1: return mitre_tags[0]
    print(f'WARN: more than one or no mitre_tags found: mitre_tags = {mitre_tags}')
    return None

def get_tactics(technique):
    kill_chain_phases = technique.get('kill_chain_phases', [])
    tactics = list(map(lambda x: x.get('phase_name'), 
                       filter(lambda x: x.get('kill_chain_name') == 'mitre-attack', kill_chain_phases)
                      ))
    return tactics

def define_tag_tactic_mapping(techniques):
    tag2tactics = {}
    for it,technique in enumerate(techniques):
        mitre_tag = get_mitre_tag(technique)
        tactics = get_tactics(technique)
        tag2tactics[mitre_tag] = tactics
    return tag2tactics

# ----- create dictionary with mapping MITRE tag → MITRE tactics
techniques = download_MITRE_data()
tag2tactics = define_tag_tactic_mapping(techniques)
print(f'Found {len(tag2tactics)} different MITRE tags and created mapping to corresponding tactics in the dictionary tag2tactics.')<jupyter_output>Enterprise ATT&CK: 95ecc380-afe9-11e4-9b6c-751b66dd541e
PRE-ATT&CK: 062767bd-02d2-4b72-84ba-56caef0f8658
Mobile ATT&CK: 2f669986-b40b-4423-b720-4396ca6a462b
ICS ATT&CK: 02c3ef24-9cd4-48f3-a99f-b74ce24f1d34<jupyter_text>Please note that a MITRE tag can be associated with multiple tactics, so this is a one-to-many relationship.  Load alert data As recreation of our test environment is beyond this article, we provide the alert data that we collected during the run of an attack here as JSON files that we load in this step into a combined list `alert_data`.<jupyter_code># --- loading the alert data collected in our test environment while running the scripted attack
data_dir = './data/mitre-tag-timeline/'
data_files = ['elastic_snort.json', 'elastic_elastalert.json', 'elastic_sysflow.json']

alert_data = []
for data_file in data_files:
    # --- load the JSON file with the alerts
    with open(data_dir + data_file, 'r') as inp:
        _data = json.load(inp)
    # --- add selected parts to alert_data
    alert_data.extend(list(map(lambda x: {'tags': x['tags'], 
                             'event.start': x['event'].get('start', None), 
                             'event.created': x['event'].get('created', None), 
                             'event.reason': x['event']['reason'], 
                             'event.severity': x['event']['severity'], 
                             'event.category': x['event']['category'], 
                             'event.action': x['event']['action'], 
                             'source_file': data_file}, _data)))<jupyter_output><empty_output><jupyter_text>Combine alert data with MITRE tactics using the mapping With the observed alerts in `alert_data`, we use our previously generated `tag2tactics` dictionary to add the tactic(s) corresponding the MITRE tags to the alerts.As each alert can have multiple tags, and each MITRE tag in turn can belong to multiple tactics, we have to duplicate alert info accordingly to account for multiple related tactics.The enriched data is then converted to a pandas DataFrame `alert_data_tactics` while fixing the timestamps coming from different columns.<jupyter_code># --- use tags to add tactics: as there can be multiple tags per alert as well as multiple tactics for a tag, create copies of the alert as required
alert_data_tactics = []
for alert in alert_data:
    tags = alert['tags']
    all_tactics = set()
    for tag in tags:
        tactics = tag2tactics[tag.replace('mitre:', '').upper()]
        for tactic in tactics:
            all_tactics.add(tactic)
    for tactic in all_tactics:
        alert['tactic'] = tactic
        alert_data_tactics.append(alert.copy())
# --- convert data to pandas DataFrame
alert_data_tactics = pd.DataFrame(alert_data_tactics)
# --- depending on source file, the timestamps are in different columns ('event.start' vs 'event.created')
#     create a fixed 'timestamp' column instead 
alert_data_tactics['timestamp'] = alert_data_tactics.apply(lambda row: datetime.datetime.fromisoformat(row['event.start'][:23]) if row['event.start'] else datetime.datetime.fromisoformat(row['event.created'][:23]), axis=1)
del alert_data_tactics['event.start']
del alert_data_tactics['event.created']
alert_data_tactics = alert_data_tactics.sort_values('timestamp').reset_index(drop=True)<jupyter_output><empty_output><jupyter_text>At this point, the list of alerts with MITRE tactics contains 37 entries:<jupyter_code>alert_data_tactics<jupyter_output><empty_output><jupyter_text>Draw timeline of observed alerts Now we finally show the observed alerts as markers over the timeline of the observation with the categorical y-axis being the tactic the alert belongs to.<jupyter_code># --- define a useful coloring of the markers
tactic2color = {
    'Discovery':pl.colors.qualitative.D3[0],
    'Initial Access':pl.colors.qualitative.D3[1],
    'Execution':pl.colors.qualitative.D3[2],
    'Privilege Escalation':pl.colors.qualitative.D3[3],
    'Defense Evasion':pl.colors.qualitative.D3[4],
    'Exfiltration':pl.colors.qualitative.D3[5],
    'Credential Access':pl.colors.qualitative.D3[6],
    'Persistence':pl.colors.qualitative.D3[7],
    'Lateral Movement':pl.colors.qualitative.D3[8],
    'discovery':pl.colors.qualitative.D3[0],
    'initial-access':pl.colors.qualitative.D3[1],
    'execution':pl.colors.qualitative.D3[2],
    'privilege-escalation':pl.colors.qualitative.D3[3],
    'defense-evasion':pl.colors.qualitative.D3[4],
    'exfiltration':pl.colors.qualitative.D3[5],
    'credential-access':pl.colors.qualitative.D3[6],
    'persistence':pl.colors.qualitative.D3[7],
    'lateral-movement':pl.colors.qualitative.D3[8],
}

# --- show the alerts as a scatter plot
fig = go.Figure(go.Scatter(x=alert_data_tactics['timestamp'], y=alert_data_tactics['tactic'], 
                           mode='markers', marker_size=18, 
                           marker_color=list(map(lambda x: tactic2color[x], alert_data_tactics['tactic'])), 
                           marker_line=dict(color = 'rgb(255, 255, 255)', width = 1),
                           text=alert_data_tactics['event.reason']))
fig.update_layout(height=800)
fig.update_layout(xaxis_title="time",
                  yaxis_title="MITRE tactic")

# --- annotate the alerts with the 'event.reason' coming from the monitoring sensors
fig.update_layout(annotations=[
            go.layout.Annotation(x=row['timestamp'], y=row['tactic'], xref="x", yref="y",
            text=row['event.reason'],
            showarrow=False, xanchor='left', yanchor='bottom', textangle=-27) for i,row in alert_data_tactics.iterrows()])
fig.show()<jupyter_output><empty_output><jupyter_text>In this view we can see a progression from discovery to access and execution, with a potential privilege escalation. We also find in the later phase some failed attempts to log into a database followed by a large data transfer - indicating a potential successful break-in with data exfiltration. Another round of discovery operations precedes more alerts with a similar sequence of access-escalation-execution and alerts tagged as lateral movement and persistence. With that high level view, looking into the details of selected alerts will confirm that the first set of alerts were related to the linux machines in our setup, whereas the later alerts come from the monitoring of our Windows machine - confirming the lateral movement that is hinted by the timeline of the tactics.In the next section, we compare this timeline with the ground truth information of our scripted attack.  Compare timeline with ground truth information As the attack itself is scripted, we know exactly what happens during this attack operation. We can use this ground truth to annotate the timeline and compare with the observed alerts.<jupyter_code># --- define the ground truth data as observed in the log of our attack script
ground_truth_attack_data = [
    (datetime.datetime.fromisoformat('2021-11-29T10:32:30.387+01:00'), "Network scans with nmap", 0), 
    (datetime.datetime.fromisoformat('2021-11-29T10:32:31.554+01:00'), "use of SQL injection attack on web shop", 0), 
    (datetime.datetime.fromisoformat('2021-11-29T10:32:40.435+01:00'), "use of SSTI vulnerability of jshop", 0), 
    (datetime.datetime.fromisoformat('2021-11-29T10:32:40.804+01:00'), "reverse shell back to attacker", 10), 
    (datetime.datetime.fromisoformat('2021-11-29T10:32:40.979+01:00'), "sequence of discovery commands by attacker", 20), 
    (datetime.datetime.fromisoformat('2021-11-29T10:33:47.687+01:00'), "preparation of downloaded attack script", 0), 
    (datetime.datetime.fromisoformat('2021-11-29T10:34:01.211+01:00'), "attempted (failed) login attemtps to MySQL server", 0), 
    (datetime.datetime.fromisoformat('2021-11-29T10:34:01.637+01:00'), "succeeds login MySQL server, exfiltrating data", 10), 
    (datetime.datetime.fromisoformat('2021-11-29T10:34:33.152+01:00'), "lateral movement: attacker uses SQL injection", 0), 
    (datetime.datetime.fromisoformat('2021-11-29T10:34:34.730+01:00'), "SSTI attack on Windows server", 0), 
    (datetime.datetime.fromisoformat('2021-11-29T10:34:35.014+01:00'), "attacker creates backdoor user", 10), 
    (datetime.datetime.fromisoformat('2021-11-29T10:34:35.031+01:00'), "attacker installs OpenSSH", 20)
]
# --- show the alerts as a scatter plot
fig = go.Figure(go.Scatter(x=alert_data_tactics['timestamp'], y=alert_data_tactics['tactic'], 
                           mode='markers', marker_size=18, 
                           marker_color=list(map(lambda x: tactic2color[x], alert_data_tactics['tactic'])), 
                           marker_line=dict(color = 'rgb(255, 255, 255)', width = 1),
                           text=alert_data_tactics['event.reason']))
fig.update_layout(height=1000)
fig.update_layout(xaxis_title="time",
                  yaxis_title="MITRE tactic")
fig.update_layout(margin=dict(l=20, r=20, t=300, b=20))

# --- annotate the alerts with the 'event.reason' coming from the monitoring sensors
annotations = [
            go.layout.Annotation(x=row['timestamp'], y=row['tactic'], xref="x", yref="y",
            text=row['event.reason'],
            showarrow=False, xanchor='left', yanchor='bottom', textangle=-27) for i,row in alert_data_tactics.iterrows()]

# --- add annotations from ground truth of known attack for comparison
for gt in ground_truth_attack_data:
    fig.add_shape(dict(type="line", x0=gt[0], y0=0.1, x1=gt[0], y1=0.88, xref='x', yref='paper', line=dict(color="darkblue", width=1), layer='below'))
annotations_ground_truth = [
        go.layout.Annotation(x=gt[0], y=0.88, xref="x", yref="paper", ax=gt[2],
                            text=gt[1], arrowcolor='darkblue', font_color='darkblue',
            showarrow=True, xanchor='left', textangle=-65) for gt in ground_truth_attack_data
    ]
fig.update_layout(annotations=annotations + annotations_ground_truth)
fig.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09074971640713622, 'fraction_numerical': 0.029132721460245436, 'mean_word_length': 4.270380434782608, 'pattern_counts': {'":': 1, '<': 24, '<?xml version=': 0, '>': 24, 'https://': 15, 'lorem ipsum': 0, 'www.': 3, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '110826', 'license': 'Apache-2.0', 'n_tokens_mistral': 5881, 'n_tokens_neox': 5377, 'n_words': 2035, 'path': 'pynb/notebooks/MitreTagTimeline/MitreTagTimeline.ipynb', 'repo_name': 'sthagen/sysflow-telemetry-sf-apis', 'size': '19394'}