[FR] Update utility path computation to use pathlib (#3699)

* update

* Updated to pathlib

* Linting

* Add string cast where needed

* Add additional string conversion as needed

* Str conversions to support eql lib

* Attack typo

* Typo in test script

* Updated for more pathlib

* Linting

* Update to convert string to path object

* Fix typo

Author: eric-forte-elastic

detection_rules/__main__.py

@@ -5,18 +5,19 @@
 
 # coding=utf-8
 """Shell for detection-rules."""
-import os
 import sys
+from pathlib import Path
 
 import click
 
 assert (3, 12) <= sys.version_info < (4, 0), "Only Python 3.12+ supported"
 
+
 from .main import root  # noqa: E402
 
-CURR_DIR = os.path.dirname(os.path.abspath(__file__))
-CLI_DIR = os.path.dirname(CURR_DIR)
-ROOT_DIR = os.path.dirname(CLI_DIR)
+CURR_DIR = Path(__file__).resolve().parent
+CLI_DIR = CURR_DIR.parent
+ROOT_DIR = CLI_DIR.parent
 
 BANNER = r"""
 █▀▀▄ ▄▄▄ ▄▄▄ ▄▄▄ ▄▄▄ ▄▄▄ ▄▄▄ ▄▄▄ ▄   ▄      █▀▀▄ ▄  ▄ ▄   ▄▄▄ ▄▄▄

detection_rules/attack.py

@@ -17,8 +17,8 @@
 from .utils import cached, clear_caches, get_etc_path, get_etc_glob_path, read_gzip, gzip_compress
 
 PLATFORMS = ['Windows', 'macOS', 'Linux']
-CROSSWALK_FILE = Path(get_etc_path('attack-crosswalk.json'))
-TECHNIQUES_REDIRECT_FILE = Path(get_etc_path('attack-technique-redirects.json'))
+CROSSWALK_FILE = get_etc_path('attack-crosswalk.json')
+TECHNIQUES_REDIRECT_FILE = get_etc_path('attack-technique-redirects.json')
 
 tactics_map = {}
 
@@ -28,17 +28,17 @@ def load_techniques_redirect() -> dict:
     return json.loads(TECHNIQUES_REDIRECT_FILE.read_text())['mapping']
 
 
-def get_attack_file_path() -> str:
+def get_attack_file_path() -> Path:
     pattern = 'attack-v*.json.gz'
     attack_file = get_etc_glob_path(pattern)
     if len(attack_file) < 1:
         raise FileNotFoundError(f'Missing required {pattern} file')
     elif len(attack_file) != 1:
         raise FileExistsError(f'Multiple files found with {pattern} pattern. Only one is allowed')
-    return attack_file[0]
+    return Path(attack_file[0])
 
 
-_, _attack_path_base = get_attack_file_path().split('-v')
+_, _attack_path_base = str(get_attack_file_path()).split('-v')
 _ext_length = len('.json.gz')
 CURRENT_ATTACK_VERSION = _attack_path_base[:-_ext_length]
 
@@ -98,7 +98,7 @@ def load_attack_gz() -> dict:
 
 def refresh_attack_data(save=True) -> (Optional[dict], Optional[bytes]):
     """Refresh ATT&CK data from Mitre."""
-    attack_path = Path(get_attack_file_path())
+    attack_path = get_attack_file_path()
     filename, _, _ = attack_path.name.rsplit('.', 2)
 
     def get_version_from_tag(name, pattern='att&ck-v'):
@@ -126,7 +126,7 @@ def get_version_from_tag(name, pattern='att&ck-v'):
     compressed = gzip_compress(json.dumps(attack_data, sort_keys=True))
 
     if save:
-        new_path = Path(get_etc_path(f'attack-v{latest_version}.json.gz'))
+        new_path = get_etc_path(f'attack-v{latest_version}.json.gz')
         new_path.write_bytes(compressed)
         attack_path.unlink()
         print(f'Replaced file: {attack_path} with {new_path}')

detection_rules/devtools.py

@@ -90,7 +90,7 @@ def dev_group():
 def build_release(config_file, update_version_lock: bool, generate_navigator: bool, generate_docs: str,
                   update_message: str, release=None, verbose=True):
     """Assemble all the rules into Kibana-ready release files."""
-    config = load_dump(config_file)['package']
+    config = load_dump(str(config_file))['package']
     registry_data = config['registry_data']
 
     if generate_navigator:
@@ -313,7 +313,7 @@ def prune_staging_area(target_stack_version: str, dry_run: bool, exception_list:
             continue
 
         # it's a change to a rule file, load it and check the version
-        if str(change.path.absolute()).startswith(RULES_DIR) and change.path.suffix == ".toml":
+        if str(change.path.absolute()).startswith(str(RULES_DIR)) and change.path.suffix == ".toml":
             # bypass TOML validation in case there were schema changes
             dict_contents = RuleCollection.deserialize_toml_string(change.read())
             min_stack_version: Optional[str] = dict_contents.get("metadata", {}).get("min_stack_version")
@@ -441,7 +441,7 @@ def integrations_pr(ctx: click.Context, local_repo: str, token: str, draft: bool
     stack_version = Package.load_configs()["name"]
     package_version = Package.load_configs()["registry_data"]["version"]
 
-    release_dir = Path(RELEASE_DIR) / stack_version / "fleet" / package_version
+    release_dir = RELEASE_DIR / stack_version / "fleet" / package_version
     message = f"[Security Rules] Update security rules package to v{package_version}"
 
     if not release_dir.exists():
@@ -581,7 +581,7 @@ def license_check(ctx, ignore_directory):
     """Check that all code files contain a valid license."""
     ignore_directory += ("env",)
     failed = False
-    base_path = Path(get_path())
+    base_path = get_path()
 
     for path in base_path.rglob('*.py'):
         relative_path = path.relative_to(base_path)
@@ -622,7 +622,7 @@ def test_version_lock(branches: tuple, remote: str):
 
     finally:
         diff = git('--no-pager', 'diff', get_etc_path('version.lock.json'))
-        outfile = Path(get_path()).joinpath('lock-diff.txt')
+        outfile = get_path() / 'lock-diff.txt'
         outfile.write_text(diff)
         click.echo(f'diff saved to {outfile}')
 
@@ -740,7 +740,7 @@ def deprecate_rule(ctx: click.Context, rule_file: Path):
                                    deprecation_date=today,
                                    maturity='deprecated')
     contents = dataclasses.replace(rule.contents, metadata=new_meta)
-    new_rule = TOMLRule(contents=contents, path=Path(deprecated_path))
+    new_rule = TOMLRule(contents=contents, path=deprecated_path)
     new_rule.save_toml()
 
     # remove the old rule

detection_rules/ecs.py

@@ -9,7 +9,6 @@
 import json
 import os
 import shutil
-from pathlib import Path
 
 import eql
 import eql.types
@@ -88,7 +87,7 @@ def get_max_version(include_master=False):
     versions = get_schema_map().keys()
 
     if include_master and any([v.startswith('master') for v in versions]):
-        return list(Path(ECS_SCHEMAS_DIR).glob('master*'))[0].name
+        return list(ECS_SCHEMAS_DIR.glob('master*'))[0].name
 
     return str(max([Version.parse(v) for v in versions if not v.startswith('master')]))
 
@@ -303,9 +302,9 @@ def download_endpoint_schemas(target: str, overwrite: bool = True) -> None:
             flattened[f"{root_name}.{f['name']}"] = f['type']
 
     # save schema to disk
-    Path(ENDPOINT_SCHEMAS_DIR).mkdir(parents=True, exist_ok=True)
+    ENDPOINT_SCHEMAS_DIR.mkdir(parents=True, exist_ok=True)
     compressed = gzip_compress(json.dumps(flattened, sort_keys=True, cls=DateTimeEncoder))
-    new_path = Path(ENDPOINT_SCHEMAS_DIR) / f"endpoint_{target}.json.gz"
+    new_path = ENDPOINT_SCHEMAS_DIR / f"endpoint_{target}.json.gz"
     if overwrite:
         shutil.rmtree(new_path, ignore_errors=True)
     with open(new_path, 'wb') as f:

detection_rules/endgame.py

@@ -7,13 +7,12 @@
 import json
 import shutil
 import sys
-from pathlib import Path
 
 import eql
 
 from .utils import ETC_DIR, DateTimeEncoder, cached, gzip_compress, read_gzip
 
-ENDGAME_SCHEMA_DIR = Path(ETC_DIR) / "endgame_schemas"
+ENDGAME_SCHEMA_DIR = ETC_DIR / "endgame_schemas"
 
 
 class EndgameSchemaManager:

detection_rules/integrations.py

@@ -25,8 +25,8 @@
 from .utils import cached, get_etc_path, read_gzip, unzip
 from .schemas import definitions
 
-MANIFEST_FILE_PATH = Path(get_etc_path('integration-manifests.json.gz'))
-SCHEMA_FILE_PATH = Path(get_etc_path('integration-schemas.json.gz'))
+MANIFEST_FILE_PATH = get_etc_path('integration-manifests.json.gz')
+SCHEMA_FILE_PATH = get_etc_path('integration-schemas.json.gz')
 _notified_integrations = set()
 
 

detection_rules/main.py

@@ -79,7 +79,7 @@ def generate_rules_index(ctx: click.Context, query, overwrite, save_files=True):
     bulk_upload_docs, importable_rules_docs = package.create_bulk_index_body()
 
     if save_files:
-        path = Path(get_path('enriched-rule-indexes', package_hash))
+        path = get_path('enriched-rule-indexes', package_hash)
         path.mkdir(parents=True, exist_ok=overwrite)
         bulk_upload_docs.dump(path.joinpath('enriched-rules-index-uploadable.ndjson'), sort_keys=True)
         importable_rules_docs.dump(path.joinpath('enriched-rules-index-importable.ndjson'), sort_keys=True)
@@ -431,7 +431,7 @@ def create_dnstwist_index(ctx: click.Context, input_file: click.Path):
     es_client: Elasticsearch = ctx.obj['es']
 
     click.echo(f'Attempting to load dnstwist data from {input_file}')
-    dnstwist_data: dict = load_dump(input_file)
+    dnstwist_data: dict = load_dump(str(input_file))
     click.echo(f'{len(dnstwist_data)} records loaded')
 
     original_domain = next(r['domain-name'] for r in dnstwist_data if r.get('fuzzer', '') == 'original*')
@@ -496,10 +496,10 @@ def create_dnstwist_index(ctx: click.Context, input_file: click.Path):
 @click.argument('author')
 def prep_rule(author: str):
     """Prep the detection threat match rule for dnstwist data with a rule_id and author."""
-    rule_template_file = Path(get_etc_path('rule_template_typosquatting_domain.json'))
+    rule_template_file = get_etc_path('rule_template_typosquatting_domain.json')
     template_rule = json.loads(rule_template_file.read_text())
     template_rule.update(author=[author], rule_id=str(uuid4()))
-    updated_rule = Path(get_path('rule_typosquatting_domain.ndjson'))
+    updated_rule = get_path('rule_typosquatting_domain.ndjson')
     updated_rule.write_text(json.dumps(template_rule, sort_keys=True))
     click.echo(f'Rule saved to: {updated_rule}. Import this to Kibana to create alerts on all dnstwist-* indexes')
     click.echo('Note: you only need to import and enable this rule one time for all dnstwist-* indexes')

detection_rules/mappings.py

@@ -4,8 +4,8 @@
 # 2.0.
 
 """RTA to rule mappings."""
-import os
 from collections import defaultdict
+from pathlib import Path
 
 from rta import get_available_tests
 
@@ -71,9 +71,9 @@ def get_rta_files(self, rta_list=None, rule_ids=None):
 
         for rta_name in rta_list:
             # rip off the extension and add .py
-            rta_name, _ = os.path.splitext(os.path.basename(rta_name))
-            rta_path = os.path.abspath(os.path.join(RTA_DIR, rta_name + ".py"))
-            if os.path.exists(rta_path):
+            rta_name = Path(rta_name).stem
+            rta_path = (RTA_DIR / rta_name).with_suffix(".py").resolve()
+            if rta_path.exists():
                 rta_files.add(rta_path)
 
         return list(sorted(rta_files))

detection_rules/misc.py

@@ -271,7 +271,7 @@ def load_current_package_version() -> str:
 
 
 def get_default_config() -> Optional[Path]:
-    return next(Path(get_path()).glob('.detection-rules-cfg.*'), None)
+    return next(get_path().glob('.detection-rules-cfg.*'), None)
 
 
 @cached

detection_rules/ml.py

@@ -27,7 +27,7 @@
 from .utils import get_path, unzip_to_dict
 
 
-ML_PATH = Path(get_path('machine-learning'))
+ML_PATH = get_path('machine-learning')
 
 
 def info_from_tag(tag: str) -> (Literal['ml'], definitions.MachineLearningType, str, int):

detection_rules/packaging.py

@@ -69,7 +69,7 @@ def filter_rule(rule: TOMLRule, config_filter: dict, exclude_fields: Optional[di
     return True
 
 
-CURRENT_RELEASE_PATH = Path(RELEASE_DIR) / load_current_package_version()
+CURRENT_RELEASE_PATH = RELEASE_DIR / load_current_package_version()
 
 
 class Package(object):
@@ -99,8 +99,8 @@ def __init__(self, rules: RuleCollection, name: str, release: Optional[bool] = F
 
     @classmethod
     def load_configs(cls):
-        """Load configs from packages.yaml."""
-        return load_etc_dump(PACKAGE_FILE)['package']
+        """Load configs from packages.yml."""
+        return load_etc_dump(str(PACKAGE_FILE))['package']
 
     @staticmethod
     def _package_kibana_notice_file(save_dir):
@@ -175,17 +175,17 @@ def get_consolidated(self, as_api=True):
 
     def save(self, verbose=True):
         """Save a package and all artifacts."""
-        save_dir = os.path.join(RELEASE_DIR, self.name)
-        rules_dir = os.path.join(save_dir, 'rules')
-        extras_dir = os.path.join(save_dir, 'extras')
+        save_dir = RELEASE_DIR / self.name
+        rules_dir = save_dir / 'rules'
+        extras_dir = save_dir / 'extras'
 
         # remove anything that existed before
         shutil.rmtree(save_dir, ignore_errors=True)
-        os.makedirs(rules_dir, exist_ok=True)
-        os.makedirs(extras_dir, exist_ok=True)
+        rules_dir.mkdir(parents=True, exist_ok=True)
+        extras_dir.mkdir(parents=True, exist_ok=True)
 
         for rule in self.rules:
-            rule.save_json(Path(rules_dir).joinpath(rule.path.name).with_suffix('.json'))
+            rule.save_json(rules_dir / Path(rule.path.name).with_suffix('.json'))
 
         self._package_kibana_notice_file(rules_dir)
         self._package_kibana_index_file(rules_dir)
@@ -195,15 +195,15 @@ def save(self, verbose=True):
             self.save_release_files(extras_dir, self.changed_ids, self.new_ids, self.removed_ids)
 
             # zip all rules only and place in extras
-            shutil.make_archive(os.path.join(extras_dir, self.name), 'zip', root_dir=os.path.dirname(rules_dir),
-                                base_dir=os.path.basename(rules_dir))
+            shutil.make_archive(extras_dir / self.name, 'zip', root_dir=rules_dir.parent, base_dir=rules_dir.name)
 
             # zip everything and place in release root
-            shutil.make_archive(os.path.join(save_dir, '{}-all'.format(self.name)), 'zip',
-                                root_dir=os.path.dirname(extras_dir), base_dir=os.path.basename(extras_dir))
+            shutil.make_archive(
+                save_dir / f"{self.name}-all", "zip", root_dir=extras_dir.parent, base_dir=extras_dir.name
+            )
 
         if verbose:
-            click.echo('Package saved to: {}'.format(save_dir))
+            click.echo(f'Package saved to: {save_dir}')
 
     def export(self, outfile, downgrade_version=None, verbose=True, skip_unsupported=False):
         """Export rules into a consolidated ndjson file."""
@@ -419,7 +419,7 @@ def _generate_registry_package(self, save_dir):
                 asset_path = rules_dir / f'{asset["id"]}.json'
             asset_path.write_text(json.dumps(asset, indent=4, sort_keys=True), encoding="utf-8")
 
-        notice_contents = Path(NOTICE_FILE).read_text()
+        notice_contents = NOTICE_FILE.read_text()
         readme_text = textwrap.dedent("""
         # Prebuilt Security Detection Rules
 

detection_rules/rule_loader.py

@@ -24,8 +24,8 @@
 from .schemas import definitions
 from .utils import cached, get_path
 
-DEFAULT_RULES_DIR = Path(get_path("rules"))
-DEFAULT_BBR_DIR = Path(get_path("rules_building_block"))
+DEFAULT_RULES_DIR = get_path("rules")
+DEFAULT_BBR_DIR = get_path("rules_building_block")
 DEFAULT_DEPRECATED_DIR = DEFAULT_RULES_DIR / '_deprecated'
 RTA_DIR = get_path("rta")
 FILE_PATTERN = r'^([a-z0-9_])+\.(json|toml)$'

detection_rules/schemas/__init__.py

@@ -4,7 +4,6 @@
 # 2.0.
 import json
 from collections import OrderedDict
-from pathlib import Path
 from typing import List, Optional
 from typing import OrderedDict as OrderedDictType
 
@@ -29,7 +28,7 @@
     "all_versions",
 )
 
-SCHEMA_DIR = Path(get_etc_path("api_schemas"))
+SCHEMA_DIR = get_etc_path("api_schemas")
 migrations = {}
 
 
@@ -54,7 +53,7 @@ def wrapper(f):
 
 @cached
 def get_schema_file(version: Version, rule_type: str) -> dict:
-    path = Path(SCHEMA_DIR) / str(version) / f"{version}.{rule_type}.json"
+    path = SCHEMA_DIR / str(version) / f"{version}.{rule_type}.json"
 
     if not path.exists():
         raise ValueError(f"Unsupported rule type {rule_type}. Unable to downgrade to {version}")