diff --git a/.flake8 b/.flake8 deleted file mode 100644 index 789f94b0..00000000 --- a/.flake8 +++ /dev/null @@ -1,13 +0,0 @@ -# E203: whitespace before ':' - black autoformatter sometimes conflicts with this one -# E501: line too long (> 79 characters) -# W503: Line break occurred before a binary operator -# D105: Missing docstring in magic method - no need to have docstrings here - -[flake8] -ignore=E203,E501,W503,D105 -docstring-convention=numpy -exclude = - tests - venv - .venv - __init__.py diff --git a/.github/workflows/lint-publish.yml b/.github/workflows/lint-publish.yml index aed9daa9..f3d84de1 100644 --- a/.github/workflows/lint-publish.yml +++ b/.github/workflows/lint-publish.yml @@ -1,65 +1,60 @@ name: Lint, test, and publish -on: [push] +on: push jobs: - lint-test: + lint-format: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 + - name: Install poetry + run: pipx install poetry + - name: Set up Python 3.11 uses: actions/setup-python@v5 with: python-version: "3.11" + cache: "poetry" - name: Install dependencies - run: | - python -m pip install --upgrade pip - pip install -r requirements-dev.txt - pip install . + run: poetry install --with=dev - - name: Lint with flake8 - run: flake8 mitreattack/ --count --exit-zero --statistics + - name: Lint with ruff + run: poetry run ruff check --exit-zero --output-format github - # should turn these back on once they take less than 10 minutes to run - # - name: Run pytest - # run: | - # cd tests - # pytest --cov=mitreattack --cov-report html + - name: Check ruff format + run: poetry run ruff format --check publish: - needs: lint-test + needs: lint-format if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags') runs-on: ubuntu-latest - # required for PyPI Trusted Publisher setup - # https://docs.pypi.org/trusted-publishers/using-a-publisher/ environment: release + # this permission is required for PyPI Trusted Publisher setup + # https://docs.pypi.org/trusted-publishers/using-a-publisher/ permissions: id-token: write steps: - uses: actions/checkout@v4 + - name: Install poetry + run: pipx install poetry + - name: Set up Python 3.11 uses: actions/setup-python@v5 with: python-version: "3.11" + cache: "poetry" - name: Install dependencies - run: | - python -m pip install --upgrade pip - pip install -r requirements-dev.txt - pip install . + run: poetry install --with=dev - name: Run pytest - run: | - cd tests - pytest --cov=mitreattack --cov-report html - cd .. + run: poetry run pytest --cov=mitreattack - name: Build package - run: python setup.py sdist bdist_wheel + run: poetry build - name: Publish package - if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags') uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/.readthedocs.yaml b/.readthedocs.yaml index 876be7dd..f30f3827 100644 --- a/.readthedocs.yaml +++ b/.readthedocs.yaml @@ -1,9 +1,12 @@ version: 2 build: - os: ubuntu-22.04 + os: ubuntu-24.04 tools: - python: "3.10" + python: "3.11" + +sphinx: + configuration: docs/conf.py python: install: diff --git a/docs/conf.py b/docs/conf.py index 66c7ef77..7418e10a 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -9,9 +9,9 @@ # -- Project information ----------------------------------------------------- project = "mitreattack-python" -copyright = "2022, The MITRE Corporation" -version = "2.0.0" -release = "2.0.0" +copyright = "2025, The MITRE Corporation" +version = "5.0.0" +release = "5.0.0" # -- General configuration --------------------------------------------------- extensions = [ diff --git a/examples/get_all_matrices.py b/examples/get_all_matrices.py index ace5bb69..94b5cfee 100644 --- a/examples/get_all_matrices.py +++ b/examples/get_all_matrices.py @@ -6,7 +6,7 @@ def main(): matrices = mitre_attack_data.get_matrices(remove_revoked_deprecated=True) - print(f"Retrieved {len(matrices)} ATT&CK matrices: { ', '.join([m.name for m in matrices]) }") + print(f"Retrieved {len(matrices)} ATT&CK matrices: {', '.join([m.name for m in matrices])}") if __name__ == "__main__": diff --git a/examples/get_objects_by_content.py b/examples/get_objects_by_content.py index fb60d458..183483d6 100644 --- a/examples/get_objects_by_content.py +++ b/examples/get_objects_by_content.py @@ -8,7 +8,7 @@ def main(): techniques = mitre_attack_data.get_objects_by_content("LSASS", "attack-pattern", remove_revoked_deprecated=True) print(f"There are {len(techniques)} techniques where 'LSASS' appears in the description.") - # retrieve all objects by the content of their description + # retrieve all objects by the content of their description objects = mitre_attack_data.get_objects_by_content("LSASS", None, remove_revoked_deprecated=True) print(f"There are a total of {len(objects)} objects where 'LSASS' appears in the description.") diff --git a/mitreattack/attackToExcel/attackToExcel.py b/mitreattack/attackToExcel/attackToExcel.py index ffdd985b..d03a61ac 100644 --- a/mitreattack/attackToExcel/attackToExcel.py +++ b/mitreattack/attackToExcel/attackToExcel.py @@ -2,7 +2,7 @@ import argparse import os -from typing import Dict, List +from typing import Dict, List, Optional import pandas as pd import requests @@ -16,7 +16,9 @@ SUB_CHARACTERS = ["\\", "/"] -def get_stix_data(domain: str, version: str = None, remote: str = None, stix_file: str = None) -> MemoryStore: +def get_stix_data( + domain: str, version: Optional[str] = None, remote: Optional[str] = None, stix_file: Optional[str] = None +) -> MemoryStore: """Download the ATT&CK STIX data for the given domain and version from MITRE/CTI (or just domain if a remote workbench is specified). Parameters @@ -105,12 +107,12 @@ def build_dataframes(src: MemoryStore, domain: str) -> Dict: "mitigations": stixToDf.mitigationsToDf(src), "matrices": stixToDf.matricesToDf(src, domain), "relationships": stixToDf.relationshipsToDf(src), - "datasources": stixToDf.datasourcesToDf(src) + "datasources": stixToDf.datasourcesToDf(src), } return df -def write_excel(dataframes: Dict, domain: str, version: str = None, output_dir: str = ".") -> List: +def write_excel(dataframes: Dict, domain: str, version: Optional[str] = None, output_dir: str = ".") -> List: """Given a set of dataframes from build_dataframes, write the ATT&CK dataset to output directory. Parameters @@ -262,11 +264,11 @@ def write_excel(dataframes: Dict, domain: str, version: str = None, output_dir: def export( domain: str = "enterprise-attack", - version: str = None, + version: Optional[str] = None, output_dir: str = ".", - remote: str = None, - stix_file: str = None, - mem_store: MemoryStore = None, + remote: Optional[str] = None, + stix_file: Optional[str] = None, + mem_store: Optional[MemoryStore] = None, ): """Download ATT&CK data from MITRE/CTI and convert it to Excel spreadsheets. @@ -298,6 +300,8 @@ def export( ------ TypeError Raised when missing exactly one of `remote`, `stix_file`, or `mem_store`. + ValueError + Raised when `mem_store` fails to load. """ if ( (remote and stix_file and mem_store) @@ -312,6 +316,9 @@ def export( if get_stix_from_github or remote or stix_file: mem_store = get_stix_data(domain=domain, version=version, remote=remote, stix_file=stix_file) + if mem_store is None: + raise ValueError("`mem_store` is empty - this should not be possible!") + logger.info(f"************ Exporting {domain} to Excel ************") # build dataframes diff --git a/mitreattack/attackToExcel/stixToDf.py b/mitreattack/attackToExcel/stixToDf.py index b7d5fd2a..e3c6732e 100644 --- a/mitreattack/attackToExcel/stixToDf.py +++ b/mitreattack/attackToExcel/stixToDf.py @@ -14,6 +14,7 @@ from mitreattack.constants import MITRE_ATTACK_ID_SOURCE_NAMES, PLATFORMS_LOOKUP from mitreattack.stix20 import MitreAttackData + def remove_revoked_deprecated(stix_objects): """Remove any revoked or deprecated objects from queries made to the data source.""" # Note we use .get() because the property may not be present in the JSON data. The default is False @@ -132,6 +133,8 @@ def techniquesToDf(src, domain): if subtechnique: subtechnique_of = all_sub_techniques.query([Filter("source_ref", "=", technique["id"])])[0] parent = src.get(subtechnique_of["target_ref"]) + else: + parent = None # base STIX properties row = parseBaseStix(technique) @@ -164,7 +167,7 @@ def techniquesToDf(src, domain): # domain specific fields -- enterprise if domain == "enterprise-attack": row["is sub-technique"] = subtechnique - if subtechnique: + if subtechnique and parent is not None: row["name"] = f"{parent['name']}: {technique['name']}" row["sub-technique of"] = parent["external_references"][0]["external_id"] diff --git a/mitreattack/collections/collection_to_index.py b/mitreattack/collections/collection_to_index.py index 89db38ad..08483743 100644 --- a/mitreattack/collections/collection_to_index.py +++ b/mitreattack/collections/collection_to_index.py @@ -75,12 +75,12 @@ def generate_index(name, description, root_url, files=None, folders=None, sets=N ) cleaned_bundles.append(potentially_valid_bundle) else: - print(f"cannot use bundle {potentially_valid_bundle.id} due to lack of collection object") + print(f"cannot use bundle {potentially_valid_bundle['id']} due to lack of collection object") else: - print(f"cannot use bundle {potentially_valid_bundle.id} due to lack of collection object") + print(f"cannot use bundle {potentially_valid_bundle['id']} due to lack of collection object") - index_created = None - index_modified = None + index_created = isoparse("9999-12-31T23:59:59.999Z") + index_modified = isoparse("9999-12-31T23:59:59.999Z") collections = {} # STIX ID -> collection object if files: diff --git a/mitreattack/diffStix/changelog_helper.py b/mitreattack/diffStix/changelog_helper.py index ff86e9c5..1e3b2923 100644 --- a/mitreattack/diffStix/changelog_helper.py +++ b/mitreattack/diffStix/changelog_helper.py @@ -497,15 +497,15 @@ def find_technique_detection_changes(self, new_stix_obj: dict, domain: str): old_datasource_id = old_datacomponent["x_mitre_data_source_ref"] old_datasource = all_old_domain_datasources[old_datasource_id] old_datasource_attack_id = get_attack_id(stix_obj=old_datasource) - old_datacomponent_detections[ - old_sourceref_id - ] = f"{old_datasource_attack_id}: {old_datasource['name']} ({old_datacomponent['name']})" + old_datacomponent_detections[old_sourceref_id] = ( + f"{old_datasource_attack_id}: {old_datasource['name']} ({old_datacomponent['name']})" + ) if old_sourceref_id in all_old_domain_detectionstrategies: old_detectionstrategy = all_old_domain_detectionstrategies[old_sourceref_id] old_detectionstrategy_attack_id = get_attack_id(stix_obj=old_detectionstrategy) - old_detectionstrategy_detections[ - old_sourceref_id - ] = f"{old_detectionstrategy_attack_id}: {old_detectionstrategy['name']}" + old_detectionstrategy_detections[old_sourceref_id] = ( + f"{old_detectionstrategy_attack_id}: {old_detectionstrategy['name']}" + ) for _, detection_relationship in self.data["new"][domain]["relationships"]["detections"].items(): if detection_relationship.get("x_mitre_deprecated") or detection_relationship.get("revoked"): @@ -517,34 +517,52 @@ def find_technique_detection_changes(self, new_stix_obj: dict, domain: str): new_datasource_id = new_datacomponent["x_mitre_data_source_ref"] new_datasource = all_new_domain_datasources[new_datasource_id] new_datasource_attack_id = get_attack_id(stix_obj=new_datasource) - new_datacomponent_detections[ - new_sourceref_id - ] = f"{new_datasource_attack_id}: {new_datasource['name']} ({new_datacomponent['name']})" + new_datacomponent_detections[new_sourceref_id] = ( + f"{new_datasource_attack_id}: {new_datasource['name']} ({new_datacomponent['name']})" + ) if new_sourceref_id in all_new_domain_detectionstrategies: new_detectionstrategy = all_new_domain_detectionstrategies[new_sourceref_id] new_detectionstrategy_attack_id = get_attack_id(stix_obj=new_detectionstrategy) - new_detectionstrategy_detections[ - new_sourceref_id - ] = f"{new_detectionstrategy_attack_id}: {new_detectionstrategy['name']}" + new_detectionstrategy_detections[new_sourceref_id] = ( + f"{new_detectionstrategy_attack_id}: {new_detectionstrategy['name']}" + ) shared_datacomponent_detections = old_datacomponent_detections.keys() & new_datacomponent_detections.keys() brand_new_datacomponent_detections = new_datacomponent_detections.keys() - old_datacomponent_detections.keys() dropped_datacomponent_detections = old_datacomponent_detections.keys() - new_datacomponent_detections.keys() new_stix_obj["changelog_datacomponent_detections"] = { - "shared": sorted([f"{new_datacomponent_detections[stix_id]}" for stix_id in shared_datacomponent_detections]), - "new": sorted([f"{new_datacomponent_detections[stix_id]}" for stix_id in brand_new_datacomponent_detections]), - "dropped": sorted([f"{old_datacomponent_detections[stix_id]}" for stix_id in dropped_datacomponent_detections]), + "shared": sorted( + [f"{new_datacomponent_detections[stix_id]}" for stix_id in shared_datacomponent_detections] + ), + "new": sorted( + [f"{new_datacomponent_detections[stix_id]}" for stix_id in brand_new_datacomponent_detections] + ), + "dropped": sorted( + [f"{old_datacomponent_detections[stix_id]}" for stix_id in dropped_datacomponent_detections] + ), } - shared_detectionstrategy_detections = old_detectionstrategy_detections.keys() & new_detectionstrategy_detections.keys() - brand_new_detectionstrategy_detections = new_detectionstrategy_detections.keys() - old_detectionstrategy_detections.keys() - dropped_detectionstrategy_detections = old_detectionstrategy_detections.keys() - new_detectionstrategy_detections.keys() + shared_detectionstrategy_detections = ( + old_detectionstrategy_detections.keys() & new_detectionstrategy_detections.keys() + ) + brand_new_detectionstrategy_detections = ( + new_detectionstrategy_detections.keys() - old_detectionstrategy_detections.keys() + ) + dropped_detectionstrategy_detections = ( + old_detectionstrategy_detections.keys() - new_detectionstrategy_detections.keys() + ) new_stix_obj["changelog_detectionstrategy_detections"] = { - "shared": sorted([f"{new_detectionstrategy_detections[stix_id]}" for stix_id in shared_detectionstrategy_detections]), - "new": sorted([f"{new_detectionstrategy_detections[stix_id]}" for stix_id in brand_new_detectionstrategy_detections]), - "dropped": sorted([f"{old_detectionstrategy_detections[stix_id]}" for stix_id in dropped_detectionstrategy_detections]), + "shared": sorted( + [f"{new_detectionstrategy_detections[stix_id]}" for stix_id in shared_detectionstrategy_detections] + ), + "new": sorted( + [f"{new_detectionstrategy_detections[stix_id]}" for stix_id in brand_new_detectionstrategy_detections] + ), + "dropped": sorted( + [f"{old_detectionstrategy_detections[stix_id]}" for stix_id in dropped_detectionstrategy_detections] + ), } def load_domain(self, domain: str): @@ -562,7 +580,9 @@ def load_domain(self, domain: str): else: directory = self.old if datastore_version == "old" else self.new if directory is None: - raise ValueError(f"Directory path for {datastore_version} data cannot be None when not using MITRE CTI") + raise ValueError( + f"Directory path for {datastore_version} data cannot be None when not using MITRE CTI" + ) stix_file = os.path.join(directory, f"{domain}.json") attack_version = release_info.get_attack_version(domain=domain, stix_file=stix_file) @@ -1009,11 +1029,11 @@ def get_markdown_string(self) -> str: for domain in self.data["changes"][object_type]: # e.g "Enterprise" - domains += f"### {self.domain_to_domain_label[domain]}\n\n" + next_domain = f"### {self.domain_to_domain_label[domain]}\n\n" # Skip mobile section for data sources if domain == "mobile-attack" and object_type == "datasource": logger.debug("Skipping - ATT&CK for Mobile does not support data sources") - domains += "ATT&CK for Mobile does not support data sources\n\n" + next_domain += "ATT&CK for Mobile does not support data sources\n\n" continue domain_sections = "" for section, stix_objects in self.data["changes"][object_type][domain].items(): @@ -1031,10 +1051,12 @@ def get_markdown_string(self) -> str: domain_sections += f"{header}\n\n{section_items}\n" # add domain sections - domains += f"{domain_sections}" + if domain_sections != "": + domains += f"{next_domain}{domain_sections}" # e.g "techniques" - content += f"## {self.attack_type_to_title[object_type]}\n\n{domains}" + if domains != "": + content += f"## {self.attack_type_to_title[object_type]}\n\n{domains}" # Add contributors if requested by argument if self.include_contributors: @@ -1677,7 +1699,9 @@ def write_detailed_html(html_file_detailed: str, diffStix: DiffStix): if change_type in ["additions", "revocations", "deprecations", "deletions"]: if stix_object.get("description"): - lines.append(f"

Description: {markdown.markdown(stix_object['description'])}

") + lines.append( + f"

Description: {markdown.markdown(stix_object['description'])}

" + ) if change_type == "revocations": revoked_by_id = get_attack_id(stix_object["revoked_by"]) @@ -1735,7 +1759,9 @@ def write_detailed_html(html_file_detailed: str, diffStix: DiffStix): lines.append("") if stix_object.get("changelog_detectionstrategy_detections"): new_detections = stix_object["changelog_detectionstrategy_detections"].get("new") - dropped_detections = stix_object["changelog_detectionstrategy_detections"].get("dropped") + dropped_detections = stix_object["changelog_detectionstrategy_detections"].get( + "dropped" + ) if new_detections: lines.append("

New Detections (Detection Strategies -> Technique):

") lines.append("") if dropped_detections: - lines.append("

Dropped Detections (Detection Strategies -> Technique):

") + lines.append( + "

Dropped Detections (Detection Strategies -> Technique):

" + ) lines.append("