Linux support with automated detection validation

.env.example

@@ -20,6 +20,9 @@
 # ATOMIC_RED_TEAM_REPO=
 # Atomic tests directory (e.g. atomics/)
 # ATOMIC_TESTS_PATH=
+# Platform (e.g. linux , default = windows)
+# PLATFORM=linux
+
 
 # --- Splunk Connection ---
 # Splunk host (hostname or IP)
@@ -61,6 +64,13 @@ ATTACK_TIDS=T1059.001,T1087.001,T1003.001
 # VM_PASSWORD=
 # Safe directory on VM for Atomic Red Team (e.g. C:\AtomicRedTeam)
 # VM_SAFE_DIR=
+# VM SSH port (default 22, e.g. 2222 for VirtualBox NAT)
+# VM_SSH_PORT=22
+# Path to SSH private key (optional; if not set, password auth is used)
+# VM_SSH_KEY_PATH=
+# Linux only: real NIC name of the VM (e.g. enp0s3). Overrides Atomic tests'
+# default "interface" arg (often eth0/ens33) so packet-capture tests work.
+# VM_INTERFACE=
 
 # --- Atomic Red Team Paths (Windows VM; optional) ---
 # Path to Invoke-AtomicRedTeam.psd1 on the VM
@@ -69,6 +79,8 @@ ATTACK_TIDS=T1059.001,T1087.001,T1003.001
 # ATOMIC_ATOMICS_PATH=C:\AtomicRedTeam\atomics
 
 # --- Proxmox (optional; for snapshot-based lab VMs) ---
+# Set to false if not using Proxmox (e.g. VirtualBox, VMware)
+# USE_PROXMOX=true
 # Proxmox host
 # PROXMOX_HOST=
 # Proxmox user (default root)

.gitignore

@@ -13,4 +13,8 @@ data/repos/
 __pycache__/
 *.pyc
 # Logs
-*.log
+*.log
+# Local scratch/analysis scripts
+_scratch/
+# PR draft docs (not part of the repo)
+PR_MESSAGE.md

add_splunk_macros.py

@@ -0,0 +1,66 @@
+"""
+Splunk ESCU Filter Macro Installer
+
+ESCU (Enterprise Security Content Updates) detection rules reference filter macros
+(e.g. `linux_auditd_add_user_account_type_filter`) that must exist in Splunk's
+macros.conf for the searches to run without errors. By default, these macros are
+not defined, causing "macro not found" errors during detection verification.
+
+This script:
+1. Scans all ESCU detection rules for `*_filter` macro references
+2. Checks which ones are already defined in macros.conf
+3. Adds missing macros with a passthrough definition (`search *`)
+
+Usage:
+    python add_splunk_macros.py
+
+Note: Requires Splunk to be installed locally. Update SPLUNK_MACROS_CONF path
+if your Splunk installation is in a different location. Restart Splunk after running.
+"""
+
+import re
+import os
+import glob
+
+DETECTIONS_PATH = os.path.join(os.path.dirname(__file__), "data", "repos",
+                               "security_content", "detections", "endpoint")
+
+SPLUNK_MACROS_CONF = r"C:\Program Files\Splunk\etc\system\local\macros.conf"
+
+def find_filter_macros():
+    macros = set()
+    pattern = re.compile(r'`(\w+_filter)`')
+    for yml_file in glob.glob(os.path.join(DETECTIONS_PATH, "*.yml")):
+        with open(yml_file, "r", encoding="utf-8") as f:
+            for match in pattern.finditer(f.read()):
+                macros.add(match.group(1))
+    return sorted(macros)
+
+def read_existing_macros():
+    existing = set()
+    if os.path.exists(SPLUNK_MACROS_CONF):
+        with open(SPLUNK_MACROS_CONF, "r", encoding="utf-8") as f:
+            for line in f:
+                m = re.match(r'\[(\w+)\]', line.strip())
+                if m:
+                    existing.add(m.group(1))
+    return existing
+
+def main():
+    macros = find_filter_macros()
+    existing = read_existing_macros()
+    new_macros = [m for m in macros if m not in existing]
+    print(f"Found {len(macros)} filter macros in ESCU rules")
+    print(f"Already defined: {len(existing)}")
+    print(f"New to add: {len(new_macros)}")
+    if not new_macros:
+        print("Nothing to add!")
+        return
+    with open(SPLUNK_MACROS_CONF, "a", encoding="utf-8") as f:
+        for macro_name in new_macros:
+            f.write(f"\n[{macro_name}]\ndefinition = search *\n")
+    print(f"Added {len(new_macros)} macros to {SPLUNK_MACROS_CONF}")
+    print("Restart Splunk for changes to take effect.")
+
+if __name__ == "__main__":
+    main()

automation/config.py

@@ -11,6 +11,10 @@
 dotenv_path = os.path.join(PROJECT_ROOT, '.env')
 load_dotenv(dotenv_path=dotenv_path)
 
+PLATFORM =  os.getenv("PLATFORM", "windows").lower()
+
+
+
 # Repo base: default data/repos (managed by RepoManager); override via REPOS_BASE_PATH in .env
 DEPENDENCIES_PATH = os.path.join(PROJECT_ROOT, 'dependencies')
 REPOS_BASE_PATH = os.getenv("REPOS_BASE_PATH", os.path.join(PROJECT_ROOT, "data", "repos"))
@@ -50,11 +54,22 @@ def _as_bool(val: str | None, default: bool = False) -> bool:
 VM_USERNAME = os.getenv("VM_USERNAME")
 VM_PASSWORD = os.getenv("VM_PASSWORD")
 VM_SAFE_DIR = os.getenv("VM_SAFE_DIR")
-
-ATOMIC_MODULE_PATH = os.getenv("ATOMIC_MODULE_PATH", r"C:\AtomicRedTeam\invoke-atomicredteam\Invoke-AtomicRedTeam.psd1")
-ATOMIC_ATOMICS_PATH = os.getenv("ATOMIC_ATOMICS_PATH", r"C:\AtomicRedTeam\atomics")
+VM_SSH_PORT = int(os.getenv("VM_SSH_PORT", "22"))
+VM_SSH_KEY_PATH = os.getenv("VM_SSH_KEY_PATH")
+# Linux only: real network interface name of the VM (e.g. enp0s3). When set, it
+# overrides an Atomic test's default "interface" argument (often eth0/ens33),
+# which otherwise fails with "No such device" on a differently-named NIC.
+VM_INTERFACE = os.getenv("VM_INTERFACE")
+
+if PLATFORM =="windows":
+    ATOMIC_MODULE_PATH = os.getenv("ATOMIC_MODULE_PATH", r"C:\AtomicRedTeam\invoke-atomicredteam\Invoke-AtomicRedTeam.psd1")
+    ATOMIC_ATOMICS_PATH = os.getenv("ATOMIC_ATOMICS_PATH", r"C:\AtomicRedTeam\atomics")
+else:
+    ATOMIC_MODULE_PATH = None
+    ATOMIC_ATOMICS_PATH = os.getenv("ATOMIC_ATOMICS_PATH", os.path.join(PROJECT_ROOT, "data", "repos", "atomic-red-team", "atomics"))
 
 # --- Proxmox settings (from .env) ---
+USE_PROXMOX = _as_bool(os.getenv("USE_PROXMOX"), True)
 PROXMOX_HOST = os.getenv("PROXMOX_HOST")
 PROXMOX_USER = os.getenv("PROXMOX_USER", "root")
 PROXMOX_PASSWORD = os.getenv("PROXMOX_PASSWORD")
@@ -68,8 +83,15 @@ def _as_bool(val: str | None, default: bool = False) -> bool:
 SPLUNK_INDEX_WAIT_SECONDS = int(os.getenv("SPLUNK_INDEX_WAIT_SECONDS", "900"))
 # Time padding around execution window when querying Splunk (seconds)
 SPLUNK_TIME_PAD_SECONDS = int(os.getenv("SPLUNK_TIME_PAD_SECONDS", "300"))
-# Post-test wait (seconds) before powering off VM to allow UF to forward events
-POST_EXEC_FORWARD_WAIT_SECONDS = int(os.getenv("POST_EXEC_FORWARD_WAIT_SECONDS", "30"))
+# Post-test wait (seconds) before powering off VM to allow UF to forward events.
+# Linux (auditd -> UF) needs longer; Windows keeps the original 30s default so its
+# run duration and search window are unchanged.
+_default_forward_wait = "100" if PLATFORM == "linux" else "30"
+POST_EXEC_FORWARD_WAIT_SECONDS = int(os.getenv("POST_EXEC_FORWARD_WAIT_SECONDS", _default_forward_wait))
+# Linux only: extra seconds added to the ESCU search window. Many ESCU rules
+# aggregate over time (e.g. `bucket _time span=15m`) and need a window wider than
+# a single test's ~100s. Default 900s (15m) to cover the common bucket span.
+ESCU_AGG_WINDOW_SECONDS = int(os.getenv("ESCU_AGG_WINDOW_SECONDS", "900"))
 
 # --- Per-test verification settings ---
 PER_TEST_VERIFICATION = _as_bool(os.getenv("PER_TEST_VERIFICATION"), False)
@@ -86,11 +108,20 @@ def _as_bool(val: str | None, default: bool = False) -> bool:
 # --- VM command execution timeout (seconds) ---
 VM_COMMAND_TIMEOUT_SECONDS = int(os.getenv("VM_COMMAND_TIMEOUT_SECONDS", "600"))
 
-ATTACK_TIDS_DEFAULT = "T1059.001,T1087.001,T1003.001"
+if PLATFORM =="windows":
+    ATTACK_TIDS_DEFAULT = "T1059.001,T1087.001,T1003.001"
+else: 
+    ATTACK_TIDS_DEFAULT = "T1059.004,T1087.001,T1222.002"
+
 ATTACK_LIST = [t.strip().upper() for t in os.getenv("ATTACK_TIDS", ATTACK_TIDS_DEFAULT).split(",") if t.strip()]
 
 # --- Output paths ---
-# Main report: dist/ for AJAX loading by index.html
-REPORT_JSON_PATH = os.path.join(PROJECT_ROOT, "dist", "attack_rule_map.json")
+# Main report: dist/ for AJAX loading by index.html.
+# Only Linux gets a platform-suffixed file (separate artifact); Windows keeps the
+# original un-suffixed name that index.html loads, so its behaviour is unchanged.
+if PLATFORM == "linux":
+    REPORT_JSON_PATH = os.path.join(PROJECT_ROOT, "dist", "attack_rule_map_linux.json")
+else:
+    REPORT_JSON_PATH = os.path.join(PROJECT_ROOT, "dist", "attack_rule_map.json")
 # dist/ for MITRE layer and HTML (keeps root clean)
 DIST_PATH = os.path.join(PROJECT_ROOT, "dist")

automation/dependency_handler.py

@@ -6,6 +6,8 @@
 from urllib.parse import urlparse
 import urllib.request
 
+from automation import config
+
 # Define the repositories we depend on
 REPOSITORIES = {
     "sigma": "https://github.com/SigmaHQ/sigma.git",
@@ -108,13 +110,18 @@ def stage_atomic_dependencies_locally(atomic_test: dict, technique_dir: str, cac
         # If PathToAtomicsFolder reference, try to resolve file under repo and stage
         if 'PathToAtomicsFolder' in val:
             rel = val.split('PathToAtomicsFolder', 1)[1].lstrip('/\\')
-            repo_root = Path(technique_dir).parents[2]  # .../dependencies/atomic-red-team/atomics/<Txxxx>
-            full_local = repo_root / rel
+            if config.PLATFORM == "linux":
+                # On Linux the atomics live at config.ATOMIC_TESTS_PATH (.../atomics),
+                # and rel is already "<Txxxx>/src/...". parents[2] would overshoot to
+                # the repos root, so resolve directly under ATOMIC_TESTS_PATH instead.
+                full_local = Path(config.ATOMIC_TESTS_PATH) / rel
+            else:
+                full_local = Path(technique_dir).parents[2] / rel  # .../dependencies/atomic-red-team/atomics/<Txxxx>
             if full_local.exists():
                 staged.append(str(full_local))
             else:
                 # Sometimes ExternalPayloads path is up one level
-                alt = repo_root.parent / rel
+                alt = Path(technique_dir).parents[2].parent / rel
                 if alt.exists():
                     staged.append(str(alt))
 

automation/dynamic_generator.py

@@ -47,21 +47,33 @@
     "field=ScriptBlockText": "field=Message",
 }
 
-# GitHub raw URLs for rule links (master branch)
+# GitHub raw URLs for rule links.
 SIGMA_RAW_BASE = "https://raw.githubusercontent.com/SigmaHQ/sigma/master"
-ESCU_RAW_BASE = "https://raw.githubusercontent.com/splunk/security_content/master"
+# ESCU/security_content: the detections/ tree lives on the repo's DEFAULT `develop`
+# branch; `master` 404s for the whole detections/ tree (verified: develop -> HTTP 200,
+# master -> 404). Switched to `develop` for Linux only, the platform this PR covers and
+# can verify. The existing Windows path is left on `master` unchanged — the same fix
+# almost certainly applies to Windows too, but I did not change the existing Windows
+# behaviour without confirmation / a way to test it.
+if config.PLATFORM == "linux":
+    ESCU_RAW_BASE = "https://raw.githubusercontent.com/splunk/security_content/develop"
+else:
+    ESCU_RAW_BASE = "https://raw.githubusercontent.com/splunk/security_content/master"
 
 
 def _apply_cim_mapping(spl: str) -> str:
     """Apply CIM-compliant field name replacements for Sigma->Splunk compatibility."""
     if not spl or not isinstance(spl, str):
         return spl
+    if config.PLATFORM == "linux":
+        return spl
     result = spl
     for old, new in sorted(CIM_MAPPING.items(), key=lambda x: -len(x[0])):
         result = result.replace(old, new)
     return result
 
 
+
 def _normalize_sigma_spl_for_splunk(query: str) -> str:
     """
     pySigma çıktısını Splunk için normalize eder.
@@ -141,6 +153,12 @@ def collect_for_technique(self, technique_id: str) -> tuple[list, list]:
                 doc = utils.load_yaml_file(fp)
                 if not isinstance(doc, dict) or "detection" not in doc or "title" not in doc:
                     continue
+
+                if config.PLATFORM == "linux":
+                    product = doc.get("logsource", {}).get("product", "")
+                    if product != "linux":
+                        continue
+
                 tags = doc.get("tags") or []
                 if not isinstance(tags, list):
                     continue
@@ -154,7 +172,7 @@ def collect_for_technique(self, technique_id: str) -> tuple[list, list]:
                                 spl = _apply_cim_mapping(spl)
                                 spl = _normalize_sigma_spl_for_splunk(spl)
                                 rule_id = doc.get("id") or ""
-                                rel_path = os.path.relpath(fp, config.SIGMA_REPO_PATH)
+                                rel_path = os.path.relpath(fp, config.SIGMA_REPO_PATH).replace("\\", "/")
                                 rule_link = f"{SIGMA_RAW_BASE}/{rel_path}" if rel_path and not rel_path.startswith("..") else ""
                                 sigma_entries.append({
                                     "rule_name": title,
@@ -174,7 +192,14 @@ def collect_for_technique(self, technique_id: str) -> tuple[list, list]:
                 if not search or not isinstance(search, str):
                     continue
                 tags = doc.get("tags") or {}
-                attack_ids = tags.get("mitre_attack_id") or []
+                # Splunk security_content moved mitre_attack_id from under `tags`
+                # to the top level of the YAML (commit db8c7c8, 2026-05-13), which
+                # silently broke ESCU technique matching against the current repo.
+                # The lookup reads the new top-level field first, then falls back to
+                # the old `tags` location, so it is correct for both schemas and every
+                # platform (confirmed with the maintainer). Enabled for Windows too,
+                # which the current schema otherwise leaves with zero ESCU matches.
+                attack_ids = doc.get("mitre_attack_id") or tags.get("mitre_attack_id") or []
                 if isinstance(attack_ids, (str, int)):
                     attack_ids = [str(attack_ids)]
                 if not isinstance(attack_ids, list):
@@ -184,7 +209,14 @@ def collect_for_technique(self, technique_id: str) -> tuple[list, list]:
                     continue
                 title = doc.get("name") or doc.get("title") or os.path.basename(fp)
                 sanitized = self._sanitize_escu_spl(search)
-                rel_path = os.path.relpath(fp, config.ESCU_REPO_PATH)
+                if config.PLATFORM == "linux":
+                    # Pipeline runs on a Windows host, so os.path.relpath returns
+                    # backslashes; normalize to forward slashes so the GitHub raw URL
+                    # is valid (the Sigma side already does this at ~line 166). Linux
+                    # only, per the same "don't touch the Windows path" rationale above.
+                    rel_path = os.path.relpath(fp, config.ESCU_REPO_PATH).replace("\\", "/")
+                else:
+                    rel_path = os.path.relpath(fp, config.ESCU_REPO_PATH)
                 rule_link = f"{ESCU_RAW_BASE}/{rel_path}" if rel_path and not rel_path.startswith("..") else ""
                 file_path = rel_path if rel_path and not rel_path.startswith("..") else fp
                 escu_entries.append({
@@ -209,7 +241,10 @@ def run_attack(technique_id: str, test_number: int = 1) -> tuple[bool, float, fl
             logging.warning("VM not ready for %s", technique_id)
             return False, 0.0, 0.0
         start_time = time.time()
-        ok = execution_handler.run_invoke_atomic_test(technique_id, test_number)
+        if config.PLATFORM == "windows":
+            ok = execution_handler.run_invoke_atomic_test(technique_id, test_number)
+        else:
+            ok = execution_handler.run_bash_atomic_test(technique_id, test_number)
         if not ok:
             vm_handler.stop_vm()
             return False, start_time, time.time()
@@ -328,9 +363,9 @@ def run(self) -> list:
 
         for technique_id in self.technique_ids:
             tid = technique_id.upper()
-            tests = atomic_parser.get_tests_for_technique(tid, platform_filter="windows")
+            tests = atomic_parser.get_tests_for_technique(tid, platform_filter=config.PLATFORM)
             if not tests:
-                logging.info("========== Technique %s (no Windows tests) ==========", tid)
+                logging.info(f"========== Technique {tid} (no {config.PLATFORM} tests) ==========")
                 continue
 
             sigma_spl_list, escu_spl_list = self.rule_mapper.collect_for_technique(tid)
@@ -405,12 +440,22 @@ def run(self) -> list:
                     sigma_results[i]["detected"] = detected
                     sigma_results[i]["log_count"] = count
 
+                # Many ESCU rules aggregate over time (e.g. `bucket _time span=15m`
+                # | stats ...), so they need a window wider than a single test's
+                # ~100s. On Linux we widen only latest_time (push forward, never into
+                # the past), so the test's own events still fall inside a full bucket
+                # without pulling in earlier tests. Each test runs on a freshly
+                # reverted VM, so this does not cross-contaminate.
+                if config.PLATFORM == "linux":
+                    escu_latest = end_time + config.ESCU_AGG_WINDOW_SECONDS
+                else:
+                    escu_latest = end_time
                 for i, r in enumerate(escu_spl_list):
                     detected, count = verification.run(
                         r["sanitized_spl"],
                         rule_name=r["rule_name"],
                         earliest_time=start_time,
-                        latest_time=end_time,
+                        latest_time=escu_latest,
                     )
                     escu_results[i]["detected"] = detected
                     escu_results[i]["log_count"] = count