git fetch origin $(System.PullRequest.TargetBranch)

Bash

git diff --name-only --pretty= origin/main..HEAD

Bash

git --no-pager diff --unified=0 origin/main..HEAD -- detections/os/windows/download_via_certutil_exe/download_via_certutil_exe_sentinel.json

Bash

import sys
import re
import subprocess
import os
import json


def run_command(command: list) -> str:
    """Executes a shell command and returns the output."""
    try:
        # print(f"[R] Running command: {' '.join(command)}")
        output = subprocess.check_output(command, text=True, encoding="utf-8", errors="replace").strip()
        # print(f"[O] Command output:\n{'\n'.join(['\t'+line for line in output.splitlines()])}")
        return output
    except subprocess.CalledProcessError as e:
        print(f"##vso[task.logissue type=error]Error executing command: {' '.join(command)}")
        print(f"##vso[task.logissue type=error]Error message: {str(e)}")
        return ""
    except UnicodeDecodeError as e:
        print(f"##vso[task.logissue type=error]Unicode decode error: {e}")
        return ""


def get_pr_modified_files() -> list:
    """Get the pr modified files"""
    return run_command(["git", "diff", "--name-only", "--pretty=" "", "origin/main..HEAD"]).splitlines()


def get_pr_modified_file_diff_lines(file: str) -> str:
    """Get the pr modified file diff"""
    return run_command(["git", "--no-pager", "diff", "--unified=0", "origin/main..HEAD", "--", file]).splitlines()


def extract_version(line: str, version_regex) -> str:
    """Extract the version number from a line."""
    match = version_regex.search(line)
    return match.group(1) if match else None


def is_version_incremented_correctly(old_version: str, new_version: str) -> bool:
    """Check if the version is incremented correctly according to specified rules."""
    old_version_parts = old_version.split(".")
    new_version_parts = new_version.split(".")

    # Convert version parts to integers for comparison
    old_version_parts = [int(part) for part in old_version_parts]
    new_version_parts = [int(part) for part in new_version_parts]

    # Destructure the version parts for clarity
    old_major, old_minor, old_patch = old_version_parts
    new_major, new_minor, new_patch = new_version_parts

    # Check if major version is incremented
    if new_major > old_major:
        # Major version incremented, minor and patch must be reset to 0
        return new_minor == 0 and new_patch == 0

    # Check if minor version is incremented
    if new_minor > old_minor:
        # Minor version incremented, patch must be reset to 0
        return new_patch == 0

    # Check if patch version is incremented
    if new_patch > old_patch:
        # Patch version can be incremented freely if no other changes
        return new_major == old_major and new_minor == old_minor

    # If no version part is incremented correctly
    return False


def check_version_in_file(file: str, remove_pattern, add_pattern) -> bool:
    """Check if file has both added and removed version lines."""
    diff_lines = get_pr_modified_file_diff_lines(file)
    removed_version_found = False
    added_version_found = False
    version_incremented_correctly = False

    old_version = None
    new_version = None
    for line in diff_lines:
        if remove_pattern.search(line):
            removed_version_found = True
            old_version = extract_version(line, remove_pattern)
        elif add_pattern.search(line):
            added_version_found = True
            new_version = extract_version(line, add_pattern)

    if removed_version_found and added_version_found:
        print("  - Version updated.")
    else:
        print("  - Version NOT updated.")

    if old_version and new_version:
        print(f"  - {old_version} -> {new_version}")
        version_incremented_correctly = is_version_incremented_correctly(old_version, new_version)
        print(f"  - Version incremented correctly: {version_incremented_correctly}")

    return removed_version_found and added_version_found and version_incremented_correctly


def get_content_packs_for_detection(detection: str) -> list:
    detection = os.path.dirname(detection.removeprefix("detections/"))
    content_packs = []

    for filename in os.listdir("content_packs"):
        if not filename.endswith(".json"):
            continue

        full_path = os.path.join("content_packs", filename)
        with open(full_path, "r", encoding="utf-8") as f:
            try:
                data = json.load(f)
            except json.JSONDecodeError:
                print(f"Invalid JSON in {filename}")
                continue

        content_pack_detections = data.get("detections", [])
        if detection in content_pack_detections:
            content_packs.append(os.path.join("content_packs", filename))

    return content_packs


def validate_version():
    version_updated_files = []
    version_not_updated_files = []

    cp_version_pattern_remove = re.compile(r'^-\s*"version"\s*:\s*"(\d+\.\d+\.\d+)"')
    cp_version_pattern_added = re.compile(r'^\+\s*"version"\s*:\s*"(\d+\.\d+\.\d+)"')
    de_version_pattern_remove = re.compile(r"^-\s*version\s*:\s*(\d+\.\d+\.\d+)")
    de_version_pattern_added = re.compile(r"^\+\s*version\s*:\s*(\d+\.\d+\.\d+)")

    pr_modified_files = get_pr_modified_files()
    print(f"Modified Files:\n{', '.join(pr_modified_files)}")

    # Content packs first as they will be used in the detections check below
    for pr_modified_file in pr_modified_files:
        if pr_modified_file.startswith("content_pack"):
            # for content_pack check version directly in file
            print(f"Checking file: {pr_modified_file}")
            if check_version_in_file(pr_modified_file, cp_version_pattern_remove, cp_version_pattern_added):
                version_updated_files.append(pr_modified_file)
            else:
                version_not_updated_files.append(pr_modified_file)

    # Detections checks for version updates
    for pr_modified_file in pr_modified_files:
        if pr_modified_file.startswith("detections"):
            print(f"Checking file: {pr_modified_file}")
            if pr_modified_file.endswith("_meta.yml"):
                # Check version in _meta.yml file directly
                if check_version_in_file(pr_modified_file, de_version_pattern_remove, de_version_pattern_added):
                    version_updated_files.append(pr_modified_file)
                else:
                    version_not_updated_files.append(pr_modified_file)

            elif pr_modified_file.endswith("_sentinel.json"):
                # check the _meta.yml file for version changes
                base_name = pr_modified_file.removesuffix("_sentinel.json")
                meta_file = base_name + "_meta.yml"

                # Check if meta file is changed in this pr
                if meta_file in pr_modified_files:
                    if check_version_in_file(meta_file, de_version_pattern_remove, de_version_pattern_added):
                        version_updated_files.append(meta_file)
                    else:
                        version_not_updated_files.append(meta_file)
                else:
                    print(f"  - Metadata file for {pr_modified_file} not updated at all.")
                    version_not_updated_files.append(meta_file)
            else:
                pass

            # If a detection belonging to a content pack increments is modified, content pack version must be modified too.
            included_in_content_packs = get_content_packs_for_detection(pr_modified_file)
            for content_pack in included_in_content_packs:
                if content_pack not in version_updated_files:
                    print(
                        f"  - Detection {pr_modified_file} modified but version not incremented in content pack {content_pack} that references it."
                    )
                    version_not_updated_files.append(content_pack)

    if version_not_updated_files:
        print("##vso[task.logissue type=error] The following files did NOT increment their version:")
        for f in version_not_updated_files:
            print(f"##vso[task.logissue type=error]  - {f}")
        sys.exit(1)
    else:
        print("All relevant files have updated their version.")


def main():
    validate_version()


if __name__ == "__main__":
    main()

Python

name: Validate Version

trigger: none

jobs:
- job: ValidateVersion
  displayName: "Validate Version"
  steps:
    - checkout: self
      fetchDepth: 0
      persistCredentials: true
    - script: |
        echo "PR ID: $(System.PullRequest.PullRequestId)"
        echo "Source Branch: $(System.PullRequest.SourceBranch)"
        echo "Target Branch: $(System.PullRequest.TargetBranch)"
        echo "Latest Commit: $(System.PullRequest.SourceCommitId)"
        git fetch origin $(System.PullRequest.TargetBranch)
      displayName: 'Fetch Target Branch'
    - script: |
        python pipelines/scripts/validate_version.py
      displayName: 'Run Validate Version'

YAML

git log -p -S "1.2.0" -- detections\cloud\azure\azure_ad_azure_discovery_using_offensive_tools\azure_ad_azure_discovery_using_offensive_tools_meta.yml

Bash

git checkout 14693f7c8b35c977ece8f664ea28e7d1270cada3 -- detections/cloud/azure/azure_ad_azure_discovery_using_offensive_tools/*

Bash

git checkout release-2025-08-03 -- content_packs/* detections/*

Bash

# Release Notes
{%- for release, content_packs in releases.items() %}
## {{ release }}
{% for content_pack_filename, content_pack_data in content_packs.items() -%}
{%- if content_pack_data.status == 'new' or content_pack_data.status == 'updated' %}
<details>
<summary><b>{{content_pack_data.status | title}}: {{ content_pack_data.name }} ({{ content_pack_filename }}) - {{content_pack_data.version}}</b></summary>
{{ content_pack_data.detections.description }}
#### Detections:
{%- for detection in content_pack_data.detections -%}
{%- if detection.status == 'new' or detection.status == 'updated' %}
**{{ detection.status| title}}**: <a href="{{repo_url}}?path=/detections/{{detection.path}}&version=GT{{release}}">{{ detection.name }}</a> - {{ detection.version }}
{% endif -%}
{%- endfor %}
</details>
{% endif -%}
{%- endfor -%}
{%- endfor -%}

Markdown

import subprocess
import os
import json
import yaml
import argparse
from jinja2 import Environment, FileSystemLoader


def run_command(command: list) -> str:
    """Executes a shell command and returns the output."""
    try:
        # print(f"[R] Running command: {' '.join(command)}")
        output = subprocess.check_output(command, text=True, encoding="utf-8", errors="replace").strip()
        # print(f"[O] Command output:\n{'\n'.join(['\t'+line for line in output.splitlines()])}")
        return output
    except subprocess.CalledProcessError as e:
        print(f"##vso[task.logissue type=error] Error executing command: {' '.join(command)}")
        print(f"##vso[task.logissue type=error] Error message: {str(e)}")
        return ""
    except UnicodeDecodeError as e:
        print(f"##vso[task.logissue type=error] Unicode decode error: {e}")
        return ""


def get_release_tags() -> list:
    """Get the list of tags in the repo"""
    return run_command(["git", "tag", "-l", "release-*"]).splitlines()


def checkout_dirs_to_ref(ref: str, dirs: str) -> str:
    """Checkout to specific tag"""
    return run_command(["git", "checkout", ref, "--"] + dirs)


def collect_content_pack_info():
    content_packs = {}

    for filename in os.listdir("content_packs"):
        if not filename.endswith(".json"):
            continue

        full_path = os.path.join("content_packs", filename)
        with open(full_path, "r", encoding="utf-8") as f:
            try:
                data = json.load(f)
            except json.JSONDecodeError:
                print(f"Invalid JSON in {filename}")
                continue

        content_pack_info = {
            "name": data.get("name", "n/a"),
            "description": data.get("description", "n/a"),
            "version": data.get("version", "n/a"),
            "detections": [],
            "status": "n/a",
        }

        for detection_path in data.get("detections", []):
            # Build the full path to the meta.yml file
            detection_full_path = os.path.join("detections", detection_path)
            detection_file_base = os.path.basename(detection_path)
            meta_file = os.path.join(detection_full_path, f"{detection_file_base}_meta.yml")

            detection_entry = {
                "path": detection_path,
                "base_name": detection_file_base,
                "version": "n/a",
                "name": "n/a",
                "description": "n/a",
                "status": "n/a",
            }

            if os.path.exists(meta_file):
                with open(meta_file, "r", encoding="utf-8") as yf:
                    try:
                        yaml_data = yaml.safe_load(yf)
                        detection_entry["version"] = yaml_data.get("version", "n/a")
                        detection_entry["name"] = yaml_data.get("title", "n/a")
                        detection_entry["description"] = yaml_data.get("description", "n/a")
                    except yaml.YAMLError:
                        print(f"##vso[task.logissue type=error]Invalid YAML in {meta_file}")
            else:
                print(f"##vso[task.logissue type=error]Missing meta file: {meta_file}")

            content_pack_info["detections"].append(detection_entry)

        content_packs[filename] = content_pack_info

    return content_packs


def update_status(releases):
    sorted_releases = sorted(releases.keys(), reverse=True)

    for i in range(len(sorted_releases)):
        current_release = sorted_releases[i]
        current_content_packs = releases[current_release]

        if i == len(sorted_releases) - 1:  # Last release
            for (
                content_pack_filename,
                content_pack_data,
            ) in current_content_packs.items():
                content_pack_data["status"] = "new"
                for detection in content_pack_data["detections"]:
                    detection["status"] = "new"
        else:
            previous_release = sorted_releases[i + 1]
            previous_content_packs = releases[previous_release]

            for (
                content_pack_filename,
                content_pack_data,
            ) in current_content_packs.items():
                if content_pack_filename not in previous_content_packs:
                    content_pack_data["status"] = "new"
                    for detection in content_pack_data["detections"]["detections"]:
                        detection["status"] = "new"
                else:
                    previous_content_pack_data = previous_content_packs[content_pack_filename]
                    if content_pack_data["version"] != previous_content_pack_data["version"]:
                        content_pack_data["status"] = "updated"
                    else:
                        content_pack_data["status"] = "existing"

                    for detection in content_pack_data["detections"]:
                        detection_path = detection["path"]
                        previous_detection = next(
                            (d for d in previous_content_pack_data["detections"] if d["path"] == detection_path),
                            None,
                        )

                        if not previous_detection:
                            detection["status"] = "new"
                        elif detection["version"] != previous_detection["version"]:
                            detection["status"] = "updated"
                        else:
                            detection["status"] = "existing"

    return releases


def generate_release_notes(repo_url: str):
    releases = {}
    release_tags = get_release_tags()
    release_tags = sorted(release_tags, reverse=True)
    print(f"Release Tags: {', '.join(release_tags)}")
    for release_tag in release_tags:
        releases[release_tag] = {}
        print(f"Checking out to tag {release_tag}...")
        checkout_dirs_to_ref(release_tag, ["content_packs/*", "detections/*"])
        print(f"Gathering pack information for {release_tag}...")
        releases[release_tag] = collect_content_pack_info()

    checkout_dirs_to_ref("origin/main", ["content_packs/*", "detections/*"])
    releases = update_status(releases)
    print(json.dumps(releases, indent=4))
    env = Environment(loader=FileSystemLoader("pipelines/scripts/templates"))
    template = env.get_template("release_notes.jinja")
    md_content = template.render(releases=releases, repo_url=repo_url)
    md_path = os.path.join("documentation", "_release_notes.md")
    with open(md_path, "w") as f:
        f.write(md_content)


def main():
    parser = argparse.ArgumentParser(description="Process Git commits for specific changes.")
    parser.add_argument("--repo-url", type=str, required=True, help="Specify the repository url.")

    args = parser.parse_args()

    generate_release_notes(repo_url=args.repo_url)


if __name__ == "__main__":
    main()

Python

name: Change Log Generation

trigger:
  branches:
    include:
    - "main"
  paths:
    include:
    - "detections/*"
    - "content_packs/*"
    exclude:
    - ".git*"

jobs:
- job: ChangeLogGeneration
  displayName: "Change Log Generation"
  condition: eq(variables['Build.SourceBranchName'], 'main')
  steps:
    - checkout: self
      fetchDepth: 0
      persistCredentials: true
    - script: |
        git fetch origin main
        git checkout -b main origin/main
      displayName: "Fetch Branches Locally"
    - script: |
        python pipelines/scripts/change_log.py --repo-url https://dev.azure.com/<ORGANIZATION_NAME>/DAC/_git/Detection-as-Code --output-directory documentation/
      displayName: 'Run Change Log Generation'
    - bash: |
        git config --global user.email "[email protected]"
        git config --global user.name "Azure DevOps Pipeline"
        [[ $(git status --porcelain) ]] || { echo "No changes, exiting now..."; exit 0; }
        echo "Changes detected, let's commit!"
        git pull origin $(Build.SourceBranchName)
        git checkout $(Build.SourceBranchName)
        git add documentation/
        git commit --allow-empty  -m "Automated commit from pipeline - Update Change Log"
        git push origin $(Build.SourceBranchName)
      displayName: Commit Change Log to Repo

YAML