Tools to Use with ATT&CK
ATT&CK Navigator
- The ATT&CK Navigator is a web-based tool for annotating and exploring ATT&CK matrices. It can be used to visualize defensive coverage, red/blue team planning, the frequency of detected techniques, and more. + The ATT&CK Navigator is a web-based tool for annotating and exploring ATT&CK matrices. It can be used to visualize defensive coverage, red/blue team planning, the frequency of detected techniques, and more.
ATT&CK Navigator
ATT&CK Workbench
- The ATT&CK Workbench is an application allowing users to explore, create, annotate, and share extensions of + The ATT&CK Workbench is an application allowing users to explore, create, annotate, and share extensions of the ATT&CK knowledge base.
ATT&CK in STIX
- Structured Threat Information Expression (STIX™) is a language and serialization format used to - exchange cyber threat intelligence (CTI). The ATT&CK dataset is available in STIX 2.0 and STIX 2.1. - Other presentations of this dataset, including the ATT&CK Navigator and this website, are built from - the STIX data. + Structured Threat Information Expression (STIX™) is a language and serialization format used to + exchange cyber threat intelligence (CTI). The ATT&CK dataset is available in STIX 2.0 and STIX 2.1. + Other presentations of this dataset, including the ATT&CK Navigator and this website, are built from + the STIX data.
- STIX is a machine-readable format providing access to the ATT&CK knowledge base. It is the most granular - representation of the ATT&CK data, and all other representations are derived from the STIX dataset. + STIX is a machine-readable format providing access to the ATT&CK knowledge base. It is the most granular + representation of the ATT&CK data, and all other representations are derived from the STIX dataset.
- The ATT&CK STIX representation is most easily manipulated in Python using the - stix2 library. - However, because STIX is represented in JSON, other programming languages can easily interact with - the raw content. + The ATT&CK STIX representation is most easily manipulated in Python using the + stix2 library. + However, because STIX is represented in JSON, other programming languages can easily interact with + the raw content.
@@ -217,10 +217,10 @@
ATT&CK in STIX 2.0
TAXII 2.1
- The ATT&CK STIX data can also be accessed via the official ATT&CK TAXII™ server. - Trusted Automated Exchange of Intelligence Information (TAXII) is an application protocol for + The ATT&CK STIX data can also be accessed via the official ATT&CK TAXII™ server. + Trusted Automated Exchange of Intelligence Information (TAXII) is an application protocol for exchanging CTI over HTTPS. The ATT&CK TAXII server provides API access to the ATT&CK - STIX knowledge base. + STIX knowledge base.
ATT&CK in Excel
- These spreadsheets are built from the STIX dataset and provide a more human-accessible view into the - knowledge base while also supporting rudimentary querying/filtering capabilities. + These spreadsheets are built from the STIX dataset and provide a more human-accessible view into the + knowledge base while also supporting rudimentary querying/filtering capabilities.
- The Excel representation of the ATT&CK dataset below includes both master spreadsheets, containing all - object types, and individual spreadsheets for each object type. The individual type spreadsheets break - out relationships (e.g procedure examples connecting groups to techniques) into separate sheets by + The Excel representation of the ATT&CK dataset below includes both master spreadsheets, containing all + object types, and individual spreadsheets for each object type. The individual type spreadsheets break + out relationships (e.g procedure examples connecting groups to techniques) into separate sheets by relationship type, while the master spreadsheet includes all relationship types in a single sheet.
- A citations sheet can be used to look up the in-text citations which appear in some fields. For domains - that include multiple matrices, such as Mobile ATT&CK, each matrix gets its own named sheet. Unlike the - STIX dataset, objects that have been revoked or deprecated are not included in the spreadsheets. + A citations sheet can be used to look up the in-text citations which appear in some fields. For domains + that include multiple matrices, such as Mobile ATT&CK, each matrix gets its own named sheet. Unlike the + STIX dataset, objects that have been revoked or deprecated are not included in the spreadsheets.
@@ -314,12 +314,96 @@
ATT&CK in Excel
- Excel Spreadsheets representing the ATT&CK dataset: -
-ATT&CK Excel Files by Version
+ +ATT&CK in Excel
- This folder within the pip module contains an additional module for converting ATT&CK STIX data - to Excel spreadsheets. It also provides a means to access ATT&CK data as Pandas DataFrames for data analysis. + This folder within the pip module contains an additional module for converting ATT&CK STIX data + to Excel spreadsheets. It also provides a means to access ATT&CK data as Pandas DataFrames for data analysis.
ATT&CK in Excel
{% block scripts %} {{ super() }} + + + + {% endblock %} \ No newline at end of file diff --git a/modules/search/search.py b/modules/search/search.py index 7829b1ea920..777fa0d7dae 100644 --- a/modules/search/search.py +++ b/modules/search/search.py @@ -159,8 +159,7 @@ def clean(filepath): def preserve_current_version(): - """Preserve current version""" - + """Preserve current version.""" # Check for intermodule dependency if [key["module_name"] for key in modules.run_ptr if key["module_name"] == "versions"]: versions.versions.deploy_current_version() diff --git a/modules/site_config.py b/modules/site_config.py index 69167a5df16..c60f62f01ee 100644 --- a/modules/site_config.py +++ b/modules/site_config.py @@ -40,6 +40,10 @@ {"name": "pre-attack", "location": STIX_LOCATION_PRE, "alias": "PRE-ATT&CK", "deprecated": True}, ] +# Directory for attack version archives +default_archive_dir = "attack-version-archives" +ATTACK_VERSION_ARCHIVES = os.getenv("ATTACK_VERSION_ARCHIVES", default_archive_dir) + # banner for the website default_banner_message = "This is a custom instance of the MITRE ATT&CK Website. The official website can be found at attack.mitre.org." BANNER_ENABLED = os.getenv("BANNER_ENABLED", True) diff --git a/modules/software/__init__.py b/modules/software/__init__.py index e65709aeede..eb38ecb3e52 100644 --- a/modules/software/__init__.py +++ b/modules/software/__init__.py @@ -1,18 +1,11 @@ -from . import software -from . import software_config import json +from . import software, software_config + def get_priority(): return software_config.priority -# TODO commented out to resolve infinite redirect loop when run locally. Needs further testing before code removal. -# def get_redirections(): -# with open(software_config.software_redirection_location , "r", encoding="utf8") as json_redirections: -# return json.load(json_redirections) -# return [] - - def run_module(): return software.generate_software(), software_config.module_name diff --git a/modules/software/software.py b/modules/software/software.py index 3d05d55320b..e2e5f5bfd60 100644 --- a/modules/software/software.py +++ b/modules/software/software.py @@ -1,13 +1,13 @@ -from collections.abc import Iterable import json import os +from collections.abc import Iterable from loguru import logger from modules import util -from . import software_config from .. import site_config +from . import software_config def generate_software(): @@ -22,12 +22,6 @@ def generate_software(): if not os.path.isdir(software_config.software_markdown_path): os.mkdir(software_config.software_markdown_path) - # TODO resolve infinite redirect loop when run locally. Needs further testing before code removal. - # Generate redirections - util.buildhelpers.generate_redirections( - redirections_filename=software_config.software_redirection_location, redirect_md=site_config.redirect_md - ) - # Generates the markdown files to be used for page generation and verifies if a software was generated software_generated = generate_markdown_files() if not software_generated: diff --git a/modules/software/software_redirections.json b/modules/software/software_redirections.json deleted file mode 100644 index f177ca67dcd..00000000000 --- a/modules/software/software_redirections.json +++ /dev/null @@ -1,7 +0,0 @@ -[ - { - "title": "software-redirect", - "from": "software.html", - "to": "/software/" - } -] \ No newline at end of file diff --git a/modules/tactics/__init__.py b/modules/tactics/__init__.py index 015f14636a8..4e8d7cab9e0 100644 --- a/modules/tactics/__init__.py +++ b/modules/tactics/__init__.py @@ -1,7 +1,7 @@ -from . import tactics -from . import tactics_config import json +from . import tactics, tactics_config + def get_priority(): return tactics_config.priority @@ -22,12 +22,5 @@ def get_menu(): } -# TODO resolve infinite redirect loop when run locally. Needs further testing before code removal. -def get_redirections(): - with open(tactics_config.tactics_redirection_location, "r", encoding="utf8") as json_redirections: - return json.load(json_redirections) - return [] - - def run_module(): return tactics.generate_tactics(), tactics_config.module_name diff --git a/modules/tactics/tactics.py b/modules/tactics/tactics.py index 2037c9bfc6d..34f4d29cab8 100644 --- a/modules/tactics/tactics.py +++ b/modules/tactics/tactics.py @@ -9,9 +9,7 @@ def generate_tactics(): - """Responsible for verifying tactic directory and generating tactic - index markdown - """ + """Responsible for verifying tactic directory and generating tactic index markdown.""" # Create content pages directory if does not already exist util.buildhelpers.create_content_pages_dir() @@ -22,12 +20,6 @@ def generate_tactics(): if not os.path.isdir(tactics_config.tactics_markdown_path): os.mkdir(tactics_config.tactics_markdown_path) - # TODO resolve infinite redirect loop when run locally. Needs further testing before code removal. - # Generate redirections - util.buildhelpers.generate_redirections( - redirections_filename=tactics_config.tactics_redirection_location, redirect_md=site_config.redirect_md - ) - # To verify if a technique was generated tactic_generated = False @@ -62,9 +54,7 @@ def generate_tactics(): def generate_domain_markdown(domain, techniques, tactics, side_nav_data, notes, deprecated=None): - """Generate tactic index markdown for each domain and generates - shared data for tactics - """ + """Generate tactic index markdown for each domain and generates shared data for tactics.""" if tactics[domain]: # Write out the markdown file for overview of domain data = {"domain": domain.split("-")[0], "tactics_list_len": str(len(tactics[domain]))} @@ -97,7 +87,7 @@ def generate_domain_markdown(domain, techniques, tactics, side_nav_data, notes, def generate_tactic_md(tactic, domain, tactic_list, techniques, side_nav_data, notes): - """Generate markdown for given tactic""" + """Generate markdown for given tactic.""" attack_id = util.buildhelpers.get_attack_id(tactic) # Add if attack id is found @@ -152,9 +142,7 @@ def generate_tactic_md(tactic, domain, tactic_list, techniques, side_nav_data, n def get_domain_table_data(tactic_list): - """Given a tactic list, returns an array of jsons with tactic name, id - and their description - """ + """Given a tactic list, returns an array of jsons with tactic name, id and their description.""" tactic_table = [] # Set up the tactics table for a domain @@ -173,9 +161,7 @@ def get_domain_table_data(tactic_list): def get_techniques_of_tactic(tactic, techniques): - """Given a tactic and a full list of techniques, return techniques that - appear inside of tactic - """ + """Given a tactic and a full list of techniques, return techniques that appear inside of tactic.""" techniques_list = [] for technique in techniques: diff --git a/modules/tactics/tactics_redirections.json b/modules/tactics/tactics_redirections.json deleted file mode 100644 index ad5b7755d0b..00000000000 --- a/modules/tactics/tactics_redirections.json +++ /dev/null @@ -1,7 +0,0 @@ -[ - { - "title": "tactics-redirect", - "from": "tactics.html", - "to": "/tactics/" - } -] \ No newline at end of file diff --git a/modules/techniques/__init__.py b/modules/techniques/__init__.py index 1cd380f62de..6ab3867e6b0 100644 --- a/modules/techniques/__init__.py +++ b/modules/techniques/__init__.py @@ -1,7 +1,7 @@ -from . import techniques -from . import techniques_config import json +from . import techniques, techniques_config + def get_priority(): return techniques_config.priority @@ -22,12 +22,5 @@ def get_menu(): } -# TODO resolve infinite redirect loop when run locally. Needs further testing before code removal. -def get_redirections(): - with open(techniques_config.techniques_redirection_location, "r", encoding="utf8") as json_redirections: - return json.load(json_redirections) - return [] - - def run_module(): return techniques.generate_techniques(), techniques_config.module_name diff --git a/modules/techniques/techniques.py b/modules/techniques/techniques.py index 3f3fb7918be..e254d1a2cb1 100644 --- a/modules/techniques/techniques.py +++ b/modules/techniques/techniques.py @@ -21,12 +21,6 @@ def generate_techniques(): if not os.path.isdir(techniques_config.techniques_markdown_path): os.mkdir(techniques_config.techniques_markdown_path) - # TODO resolve infinite redirect loop when run locally. Needs further testing before code removal. - # Generate redirections - util.buildhelpers.generate_redirections( - redirections_filename=techniques_config.techniques_redirection_location, redirect_md=site_config.redirect_md - ) - # Write the technique index.html page with open(os.path.join(techniques_config.techniques_markdown_path, "overview.md"), "w", encoding="utf8") as md_file: md_file.write(techniques_config.technique_overview_md) diff --git a/modules/techniques/techniques_redirections.json b/modules/techniques/techniques_redirections.json deleted file mode 100644 index 4424bcaa696..00000000000 --- a/modules/techniques/techniques_redirections.json +++ /dev/null @@ -1,7 +0,0 @@ -[ - { - "title": "techniques-redirect", - "from": "techniques.html", - "to": "/techniques/" - } -] \ No newline at end of file diff --git a/modules/tests/__init__.py b/modules/tests/__init__.py index e403602f5c6..26dbffafc53 100644 --- a/modules/tests/__init__.py +++ b/modules/tests/__init__.py @@ -1,5 +1,4 @@ -from . import tests -from . import tests_config +from . import tests, tests_config def get_priority(): diff --git a/modules/tests/linkchecker.py b/modules/tests/linkchecker.py index 9e2a7f1f8fb..95ed1efde39 100644 --- a/modules/tests/linkchecker.py +++ b/modules/tests/linkchecker.py @@ -1,7 +1,10 @@ import os import re +import threading +from concurrent.futures import ThreadPoolExecutor, as_completed import requests +from loguru import logger import modules from modules import site_config @@ -10,9 +13,25 @@ # STATIC PROPERTIES -allowed_in_link_with_external_links = r"-?\w\$\.!\*'()/:" +IGNORED_LINKS = ["full-coverage.html", "macro-technique-refinement.html"] -allowed_in_link = r"-?\w\$\.!\*'()/" +ALLOWED_IN_LINK_INTERNAL = r"-?\w\$\.!\*'()/" +ALLOWED_IN_LINK_EXTERNAL = r"-?\w\$\.!\*'()/:" +LINK_POSTFIX = r']+)["\']' + +# Pre-compiled regex patterns +LINK_REGEXES = { + "href_internal": re.compile(r'href\s?=\s?["\']([' + ALLOWED_IN_LINK_INTERNAL + LINK_POSTFIX), + "href_external": re.compile(r'href\s?=\s?["\']([' + ALLOWED_IN_LINK_EXTERNAL + LINK_POSTFIX), + "src_internal": re.compile(r'src\s?=\s?["\']([' + ALLOWED_IN_LINK_INTERNAL + LINK_POSTFIX), + "src_external": re.compile(r'src\s?=\s?["\']([' + ALLOWED_IN_LINK_EXTERNAL + LINK_POSTFIX), + "cache_bust": re.compile(r"(css|js)\?\w+"), +} + +# Caches +file_exists_cache = {} +get_correct_link_cache = {} +relative_link_cache = {} links_list = {} in_use_links = {} @@ -44,8 +63,13 @@ def remove_extra_from_path(filepath): def get_correct_link(path): """Given a path, return the correct path by adding index.html or removing cache-disabling query string suffix.""" + # Use cache to avoid redundant work + if path in get_correct_link_cache: + return get_correct_link_cache[path] + # Ignore if it starts with http if path.startswith("http"): + get_correct_link_cache[path] = path return path # All paths need to start with / @@ -71,7 +95,7 @@ def get_correct_link(path): "docx", "rtf", ]: - if re.search(r"(css|js)\?\w+", sort_of_extension): + if LINK_REGEXES["cache_bust"].search(sort_of_extension): # CSS & JavaScript: check for cache-disabling query string suffix, e.g style.min.css?f8be4c06 path = path.split("?")[0] # remove suffix else: @@ -80,10 +104,10 @@ def get_correct_link(path): # serving a file add index.html to replicate webserver # functionality if not path.endswith("/"): - # logger.debug(f"does this even happen? even once? {path}") path += "/" path += "index.html" + get_correct_link_cache[path] = path return path @@ -109,7 +133,7 @@ def internal_link_test(link): web_dir = site_config.web_directory.split(site_config.subdirectory)[0] else: web_dir = site_config.web_directory - + # Get correct link path path = web_dir + get_correct_link(link) @@ -125,18 +149,26 @@ def internal_link_test(link): from_index_path = path.split("/index.html") from_index_path = from_index_path[0] + ".html" - if os.path.exists(path) or os.path.exists(to_index_path) or os.path.exists(from_index_path): - return False - else: - return True + # Use cache for file existence checks + for test_path in (path, to_index_path, from_index_path): + if test_path in file_exists_cache: + if file_exists_cache[test_path]: + return False + else: + exists = os.path.exists(test_path) + file_exists_cache[test_path] = exists + if exists: + return False + return True def check_if_relative_link(link): """Given a link, return true if it is a relative path.""" - if not link.startswith("http"): - if not link.startswith("/"): - return True - return False + if link in relative_link_cache: + return relative_link_cache[link] + result = not link.startswith("http") and not link.startswith("/") + relative_link_cache[link] = result + return result def internal_external_link_checker(filepath, html_str): @@ -154,9 +186,9 @@ def internal_external_link_checker(filepath, html_str): if ( "/versions/" in filepath ): # don't check links with data-test-ignore attribute, or live version link name, when on previous versions - linkregex = rf'{prefix}\s?=\s?["\']([{allowed_in_link_with_external_links}]+)["\'](?! ?data-test-ignore="true")(?!>live version)' + linkregex = rf'{prefix}\s?=\s?["\']([{ALLOWED_IN_LINK_EXTERNAL}]+)["\'](?! ?data-test-ignore="true")(?!>live version)' else: - linkregex = rf"{prefix}\s?=\s?[\"']([{allowed_in_link_with_external_links}]+)[\"']" + linkregex = rf"{prefix}\s?=\s?[\"']([{ALLOWED_IN_LINK_EXTERNAL}]+)[\"']" links = re.findall(linkregex, html_str) # check if link has a dest @@ -215,7 +247,7 @@ def internal_link_checker(filepath, html_str): # find all links for prefix in ["href", "src"]: - links = re.findall(rf"{prefix}\s?=\s?[\"']([{allowed_in_link}]+)[\"']", html_str) + links = re.findall(rf"{prefix}\s?=\s?[\"']([{ALLOWED_IN_LINK_INTERNAL}]+)[\"']", html_str) # check if link has a dest for link in links: # Check if link is relative path @@ -229,6 +261,11 @@ def internal_link_checker(filepath, html_str): # Get correct path link = get_correct_link(link) + for ignored_link in IGNORED_LINKS: + if ignored_link in link: + logger.debug(f"Ignoring link: {link}") + links_list[link] = None + # Check if link is in use check_if_link_in_use(filepath, link) @@ -350,34 +387,39 @@ def check_links(external_links=False): filenames = [] internal_problem = False + html_filepaths = [] for directory, _, files in os.walk(site_config.web_directory): for filename in filter(lambda f: f.endswith(".html"), files): filepath = os.path.join(directory, filename) - filenames.append(filepath) - - # Do not check previous dir with external links - if external_links and "previous" not in directory and "versions" not in directory: - report = check_links_on_page(filepath, True) + html_filepaths.append(filepath) + + # Parallelize link checking + max_workers = min(32, os.cpu_count() or 4) + lock = threading.Lock() + with ThreadPoolExecutor(max_workers=max_workers) as executor: + future_to_filepath = {} + for filepath in html_filepaths: + if external_links and "previous" not in filepath and "versions" not in filepath: + future = executor.submit(check_links_on_page, filepath, True) else: - report = check_links_on_page(filepath) - - # Set internal problem flag to true if internal - # problem is found. We want to exit out on error if an internal - # link is broken - if not internal_problem: - if report.get("internal_problem"): + future = executor.submit(check_links_on_page, filepath) + future_to_filepath[future] = filepath + + for future in as_completed(future_to_filepath): + report = future.result() + with lock: + # Set internal problem flag to true if internal link is broken + if not internal_problem and report.get("internal_problem"): internal_problem = True - - if report.get("problems"): - broken_pages.append(report) - - if report.get("relative_links"): - relative_links_report = { - "path": report["path"], - "relative_links": report["relative_links"], - } - relative_links.append(relative_links_report) + if report.get("problems"): + broken_pages.append(report) + if report.get("relative_links"): + relative_links_report = { + "path": report["path"], + "relative_links": report["relative_links"], + } + relative_links.append(relative_links_report) # Get unlinked pages list unlinked_pages = check_unlinked_pages(filenames) diff --git a/modules/tests/tests.py b/modules/tests/tests.py index 05c9595064e..3d656dfa956 100644 --- a/modules/tests/tests.py +++ b/modules/tests/tests.py @@ -12,7 +12,7 @@ def run_tests(): - """Run tests""" + """Run tests.""" error_list = [] tests = 0 @@ -118,7 +118,7 @@ def display_error_report(report_file, error_count, error_type): def check_links(external_links): - """Wrapper to check internal and/or external links""" + """Check internal and/or external links.""" # Link test TEST = "Links" @@ -171,7 +171,7 @@ def check_links(external_links): def check_citations(): - """Wrapper to check for broken citations""" + """Check for broken citations.""" TEST = "Broken Citations" logger.info(f"RUNNING {TEST}") @@ -190,7 +190,7 @@ def check_citations(): def check_size(): - """Wrapper to check output size for Github's limit""" + """Check output size for Github's limit.""" TEST = "Output Folder Size" logger.info(f"RUNNING {TEST}") diff --git a/modules/util/__init__.py b/modules/util/__init__.py index 5994d03cd71..6d2cca402b3 100644 --- a/modules/util/__init__.py +++ b/modules/util/__init__.py @@ -1,4 +1,8 @@ -from . import relationshipgetters -from . import relationshiphelpers -from . import buildhelpers -from . import stixhelpers +from . import buildhelpers, relationshipgetters, relationshiphelpers, stixhelpers + +__all__ = [ + "relationshipgetters", + "relationshiphelpers", + "buildhelpers", + "stixhelpers", +] diff --git a/modules/util/buildhelpers.py b/modules/util/buildhelpers.py index 94b952c059b..97371a7d4d0 100644 --- a/modules/util/buildhelpers.py +++ b/modules/util/buildhelpers.py @@ -1,12 +1,9 @@ import datetime import json -import math import os import re import shutil -import string import sys -import uuid import bleach from loguru import logger @@ -18,7 +15,7 @@ def timestamp(): - """This method is here to return a timestamp.""" + """Return a timestamp.""" timestamp = datetime.datetime.now().strftime("%H:%M:%S") return timestamp @@ -56,8 +53,10 @@ def get_first_last_seen_dates(obj): def get_first_last_seen_citations(obj): - """Given an object, return the first seen and last seen citations. This function generates - the descriptions/citations for the first/last seen fields.""" + """Return the first seen and last seen citations of an object. + + This function generates the descriptions/citations for the first/last seen fields. + """ data = {} if obj.get("x_mitre_first_seen_citation"): data["first_seen_citation"] = obj.get("x_mitre_first_seen_citation") @@ -75,7 +74,7 @@ def format_date_as_month_year(date): def find_index_id(ext_ref): - """This method will search for the index of the external_id in the external reference list.""" + """Search for the index of the external_id in the external reference list.""" count = 0 flag = True while count < len(ext_ref) and flag: @@ -132,8 +131,7 @@ def update_reference_list(reference_list, obj): def get_reference_set(reflist): - """This function retrieves the unique set of references in the given list of descriptions and - returns them in string format to be displayed as citations.""" + """Retrieve the unique set of references in the given list of descriptions and return them in string format to be displayed as citations.""" p = re.compile(r"\(Citation: (.*?)\)") citations = {} for c in reflist: @@ -146,7 +144,7 @@ def get_reference_set(reflist): def get_alias_data(alias_list, ext_refs): - """This function generates the Alias Description section for the pages.""" + """Generate the Alias Description section for the pages.""" if not alias_list: return [] @@ -228,7 +226,7 @@ def get_technique_table_data(tactic, techniques_list): sub_data["name"] = subtechnique["object"]["name"] sub_attack_id = get_attack_id(subtechnique["object"]) if sub_attack_id: - if not "." in sub_attack_id: + if "." not in sub_attack_id: raise Exception(f"{attack_id} subtechnique's attackID '{sub_attack_id}' is malformed") sub_data["id"] = sub_attack_id.split(".")[1] sub_data["descr"] = subtechnique["object"]["description"] @@ -296,7 +294,6 @@ def get_element_data(element): def get_side_menu_data(side_nav_title, path_prefix, elements_list, domain=None): """Responsible for generating the links that are located on the left side of pages for desktop clients.""" - elements_data = [] for element in elements_list: @@ -426,7 +423,7 @@ def technique_used_helper(technique_list, technique, reference_list, inherited=F # Check if parent ID was added by sub-technique # parent technique will be marked as not used - elif technique_list[attack_id]["technique_used"] == False: + elif not technique_list[attack_id]["technique_used"]: # Include as a technique used technique_list[attack_id]["technique_used"] = True @@ -505,7 +502,6 @@ def replace_html_chars(to_be_replaced): def get_navigator_layers(name, attack_id, obj_type, rel_type, version, techniques_used, inheritance=False): """Generate the Enterprise, Mobile, and ICS Navigator JSON layers for the given object.""" - # Generate Enterprise base layer enterprise_layer = build_base_layer("enterprise-attack", name, obj_type, rel_type, attack_id, version, inheritance) @@ -679,31 +675,6 @@ def add_platform_path(platforms): return platforms -def print_start(name): - """Given a name and a time, display current progress.""" - number_of_hyphens = 40 - name_space = 22 - - hyphens = "-" * number_of_hyphens - - sys.stdout.write(f"\r{name: <{name_space}} : {hyphens} Running...") - - sys.stdout.flush() - - -def print_end(name, start_time, end_time): - """Given a name and a time, display current progress.""" - number_of_hyphens = 40 - name_space = 22 - - hyphens = "-" * number_of_hyphens - - # spaces here because we need to overwrite the word "running" - sys.stdout.write(f"\r{name: <{name_space}} : {hyphens} {end_time - start_time:.2f}s \n") - - sys.stdout.flush() - - def filter_techniques_by_platform(tech_list, platforms): """Given a technique list and a platforms list, filter out techniques that are not part of the platforms.""" if not platforms: diff --git a/modules/util/stixhelpers.py b/modules/util/stixhelpers.py index 7437fd4e0c0..540e7bb37cf 100644 --- a/modules/util/stixhelpers.py +++ b/modules/util/stixhelpers.py @@ -5,7 +5,6 @@ import requests import stix2 -import urllib3 from loguru import logger from requests.adapters import HTTPAdapter, Retry @@ -392,7 +391,6 @@ def get_domain_resources(types): def get_stix_memory_stores(): """Read the json files for each domain and create a dict that contains the memory stores for each domain.""" - ms = {} srcs = [] @@ -401,6 +399,7 @@ def get_stix_memory_stores(): for domain in site_config.domains: stix_filename = None + logger.info(f"Loading {domain['name']} domain STIX from: {domain['location']}") # Download json from http or https stix_filename = f"{stix_output_dir}/{domain['name']}.json" @@ -410,7 +409,6 @@ def get_stix_memory_stores(): shutil.copy(domain["location"], str(stix_filename)) if os.path.exists(stix_filename): - logger.info(f"Loading STIX file from: {stix_filename}") ms[domain["name"]] = stix2.MemoryStore() ms[domain["name"]].load_from_file(stix_filename) else: diff --git a/modules/util/util_config.py b/modules/util/util_config.py index 50ca1aa43cb..10bf894164c 100644 --- a/modules/util/util_config.py +++ b/modules/util/util_config.py @@ -1,7 +1,5 @@ import shutil -import colorama - # Not found constant NOT_FOUND = -1 diff --git a/modules/versions/templates/versions.html b/modules/versions/templates/versions.html index 27884dade2f..9c1747eca97 100644 --- a/modules/versions/templates/versions.html +++ b/modules/versions/templates/versions.html @@ -1,4 +1,4 @@ -{% extends "general/two-column.html" %} +{% extends "general/two-column.html" %} {% set active_page = "resources" -%} {% set title = "Version History | MITRE ATT&CK®" -%} {% set version_history = page.data | from_json %} @@ -21,8 +21,8 @@Version History
- The overall ATT&CK catalog is versioned using a major.minor version schema.
- The bi-annual content releases listed on the updates pages increment the major version number.
+ The overall ATT&CK catalog is versioned using a major.minor version schema.
+ The ATT&CK releases listed on the updates pages increment the major version number.
The minor version number increments for our other small releases, which include typo and data corrections but not typically new content.
diff --git a/modules/versions/versions.py b/modules/versions/versions.py
index a2baea51ff4..df685b4ba1b 100644
--- a/modules/versions/versions.py
+++ b/modules/versions/versions.py
@@ -1,126 +1,87 @@
+import concurrent.futures as cf
+import functools
+import io
import json
import os
import re
import shutil
-import stat
+import subprocess
+import tarfile
+import tempfile
from datetime import datetime
+from pathlib import Path
-from git import Repo
+import requests
from loguru import logger
from modules import site_config, util
from . import versions_config
+ALLOWED = r"-?\w\$\.!\*'()/"
+SRC_RE = re.compile(rf'src=["\'](?!/versions/)([{ALLOWED}]+)["\']')
+HREF_RE = re.compile(r'href=(["\'])((?!/(versions|resources)/)/[^"\']*)\1')
+META_REDIR_RE = re.compile(r'content="0; url=(["\']?)(?!/(versions|resources)/|https?://)(/[^"\'>]*)\1')
+LIVE_BTN_RE = re.compile(rf'href=["\']/versions/v[\w-]+/([{ALLOWED}]+)["\'](.*)>[Ll]ive [Vv]ersion')
+
def generate_versions():
"""Responsible for generating the versions pages."""
# Move templates to templates directory
util.buildhelpers.move_templates(versions_config.module_name, versions_config.versions_templates_path)
- # Verify if resources directory exists
- if not os.path.isdir(site_config.resources_markdown_path):
- os.mkdir(site_config.resources_markdown_path)
-
- # Verify if resources directory exists
- if not os.path.isdir(versions_config.versions_markdown_path):
- os.mkdir(versions_config.versions_markdown_path)
-
- deploy()
-
-
-# Error handler for windows by:
-# https://stackoverflow.com/questions/2656322/shutil-rmtree-fails-on-windows-with-access-is-denied
-def onerror(func, path, exc_info):
- """
- Error handler for ``shutil.rmtree``.
-
- If the error is due to an access error (read only file)
- it attempts to add write permission and then retries.
-
- If the error is for another reason it re-raises the error.
-
- Usage : ``shutil.rmtree(path, onerror=onerror)``
- """
- try:
- if not os.access(path, os.W_OK):
- # Is the error an access error ?
- os.chmod(path, stat.S_IWUSR)
- func(path)
- except:
- raise
-
-
-# allowed characters inside of hyperlinks
-allowed_in_link = r"-?\w\$\.!\*'()/"
-
-def versionPath(version):
- """Get the path of a given version."""
- if "path" in version:
- return version["path"]
- else:
- return version["name"].split(".")[0] # parse path from name if not given explicitly
-
+ # Create resources and versions directories
+ os.makedirs(site_config.resources_markdown_path, exist_ok=True)
+ os.makedirs(versions_config.versions_markdown_path, exist_ok=True)
-def deploy():
- """Deploy previous versions to website directory."""
+ # Ensure directories exist
versions_config.prev_versions_deploy_folder = os.path.join(
site_config.web_directory, versions_config.prev_versions_path
)
-
- # TODO we probably don't need to re-clone the website here, just a git pull should be sufficient
- # delete previous copy of attack-versions
- if os.path.exists(versions_config.versions_directory):
- shutil.rmtree(versions_config.versions_directory, onerror=onerror)
- # download new version of attack-website for use in versioning
- logger.info(f"git cloning {versions_config.versions_repo} >>> {versions_config.versions_directory}")
- versions_repo = Repo.clone_from(versions_config.versions_repo, versions_config.versions_directory)
-
- # remove previously deployed previous versions
- if os.path.exists(versions_config.prev_versions_deploy_folder):
- for child in os.listdir(versions_config.prev_versions_deploy_folder):
- if os.path.isdir(os.path.join(versions_config.prev_versions_deploy_folder, child)):
- shutil.rmtree(versions_config.prev_versions_deploy_folder)
+ os.makedirs(versions_config.prev_versions_deploy_folder, exist_ok=True)
with open("data/versions.json", "r") as f:
versions = json.load(f)
- # build previous versions
- for version in versions["previous"]:
- deploy_previous_version(version, versions_repo)
-
- # build the versions page
- build_markdown(versions)
+ logger.info("Deploying preserved versions …")
+ with cf.ThreadPoolExecutor() as executor:
+ list(executor.map(deploy_previous_version, versions["previous"]))
- # Create directory if it does not exist
- if not os.path.isdir(site_config.web_directory):
- os.makedirs(site_config.web_directory)
+ build_markdown(versions=versions)
# write robots.txt to disallow crawlers
+ os.makedirs(site_config.web_directory, exist_ok=True)
with open(os.path.join(site_config.web_directory, "robots.txt"), "w", encoding="utf8") as robots:
robots.write(
- f"User-agent: *\nDisallow: {site_config.subdirectory}/previous/\nDisallow: {site_config.subdirectory}/{versions_config.prev_versions_path}/"
+ f"User-agent: *\n"
+ f"Disallow: {site_config.subdirectory}/previous/\n"
+ f"Disallow: {site_config.subdirectory}/{versions_config.prev_versions_path}/"
)
def deploy_current_version():
- """Build a permalink of the current version."""
+ """Build a permalink of the current version.
+
+ This is only called by the search module's preserve_current_version().
+ """
versions_config.prev_versions_deploy_folder = os.path.join(
site_config.web_directory, versions_config.prev_versions_path
)
with open("data/versions.json", "r") as f:
- version = json.load(f)["current"]
+ version_data = json.load(f)["current"]
- if not os.path.exists(os.path.join(versions_config.prev_versions_deploy_folder, versionPath(version))):
- os.mkdir(os.path.join(versions_config.prev_versions_deploy_folder, versionPath(version)))
+ version_path = version_data["name"].split(".")[0]
+ version_full_path = os.path.join(versions_config.prev_versions_deploy_folder, version_path)
+
+ os.makedirs(version_full_path, exist_ok=True)
for item in os.listdir(site_config.web_directory):
- # skip previous and versions directories when copying
- if item == "previous" or item == "versions":
+ # skip versions directories when copying
+ if item == "versions":
continue
# copy the current version into a preserved version
src = os.path.join(site_config.web_directory, item)
- dest = os.path.join(versions_config.prev_versions_deploy_folder, versionPath(version), item)
+ dest = os.path.join(version_full_path, item)
# copy depending on file type
if os.path.exists(dest):
print(f"error copying {src}: path {dest} already exists | {item}")
@@ -130,215 +91,352 @@ def deploy_current_version():
shutil.copy(src, dest)
# run archival scripts
- archive(version, is_current=True)
+ archive(
+ version_data=version_data,
+ version_path=version_full_path,
+ is_current=True,
+ )
-def deploy_previous_version(version, repo):
- """Build a version of the site to /prev_versions_path. version is a version from versions.json, repo is a reference to the attack-website Repo object."""
- logger.info(f"Building website for ATT&CK {version}")
- # check out the commit for that version
- repo.git.checkout(version["commit"])
- # copy over files
- ignored_stuff = shutil.ignore_patterns(
- ".git", "beta", "CNAME", "robots.txt", "previous", "previous-versions", "versions"
- )
- shutil.copytree(
- os.path.join(versions_config.versions_directory),
- os.path.join(versions_config.prev_versions_deploy_folder, versionPath(version)),
- ignore=ignored_stuff,
+def create_tar_gz(source_dir, output_path):
+ """Create a tar.gz archive from source_dir."""
+ logger.info(f"Creating tar.gz archive: {output_path}")
+ with tarfile.open(output_path, "w:gz") as tar:
+ tar.add(source_dir, arcname=".")
+ logger.info(f"Archive created: {output_path}")
+
+
+def extract_tar_gz(archive_path, dest_path):
+ """Extract a .tar.gz archive to dest_path."""
+ os.makedirs(dest_path, exist_ok=True)
+ with tarfile.open(archive_path, "r:gz") as tar:
+ tar.extractall(path=dest_path)
+
+
+def download_archive(url, local_path):
+ """Download a website archive file from GitHub. Returns True if download succeeded."""
+ try:
+ logger.info(f"Downloading website archive from {url} to {local_path}")
+ r = requests.get(url, timeout=15)
+ r.raise_for_status()
+ with open(local_path, "wb") as f:
+ f.write(r.content)
+ logger.info(f"Successfully downloaded archive: {local_path}")
+ return True
+ except Exception as ex:
+ logger.warning(f"Failed to download archive from {url}: {ex}")
+ return False
+
+
+def export_commit(commit: str, dest_path: str) -> None:
+ """Materialise `commit` into `dest_path` using `git archive`.
+
+ Equivalent to: git archive
