Panther rules: azure

from panther_azureactivity_helpers import (
    azure_activity_alert_context,
    azure_activity_success,
    extract_resource_name_from_id,
)

ACTION_GROUPS_DELETE = "MICROSOFT.INSIGHTS/ACTIONGROUPS/DELETE"


def rule(event):
    return event.get(
        "operationName", ""
    ).upper() == ACTION_GROUPS_DELETE and azure_activity_success(event)


def title(event):
    resource_id = event.get("resourceId", "")

    action_group = extract_resource_name_from_id(
        resource_id, "actionGroups", default="<UNKNOWN_ACTION_GROUP>"
    )
    return f"Azure Action Group deleted [{action_group}]"


def alert_context(event):
    context = azure_activity_alert_context(event)

    resource_id = event.get("resourceId", "")

    action_group_name = extract_resource_name_from_id(resource_id, "actionGroups", default="")
    if action_group_name:
        context["action_group_name"] = action_group_name

    resource_group = extract_resource_name_from_id(resource_id, "resourceGroups", default="")
    if resource_group:
        context["resource_group"] = resource_group

    return context

AnalysisType: rule
Filename: azure_action_groups_deleted.py
RuleID: "Azure.MonitorActivity.ActionGroups.Deleted"
DisplayName: "Azure Action Groups Deleted"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: Info
Description: >
  Detects when Azure action groups are deleted, which could disable alert notifications to security teams to prevent incident response.
Reports:
  MITRE ATT&CK:
    - TA0005:T1562.008 # Defense Evasion: Impair Defenses - Disable Cloud Logs
Tags:
  - Defense Evasion
  - Impair Defenses
  - Disable Cloud Logs
Runbook: |
  1. Query Azure Monitor Activity logs for all action group and notification operations by the callerIpAddress in the 24 hours before and after this alert to identify if multiple notification channels are being deleted
  2. Check if the source IP is associated with known cloud providers, VPN services, or corporate network ranges using threat intelligence
  3. Search for other Azure alert rule deletions or monitoring infrastructure changes from the same user or IP in the past 7 days to assess if this is part of a defense evasion campaign
Reference: https://learn.microsoft.com/en-us/azure/azure-monitor/alerts/action-groups
SummaryAttributes:
  - resourceId
  - callerIpAddress
  - correlationId

from panther_azureactivity_helpers import azure_activity_alert_context

ADVISOR_RECOMMENDATION_OPERATION = "MICROSOFT.ADVISOR/RECOMMENDATIONS/AVAILABLE/ACTION"
RECOMMENDATION_CATEGORY = "Recommendation"
SECURITY_CATEGORY = "Security"


def rule(event):
    return all(
        [
            event.get("operationName", "").upper() == ADVISOR_RECOMMENDATION_OPERATION,
            event.get("category", "") == RECOMMENDATION_CATEGORY,
            event.deep_get("properties", "recommendationCategory") == SECURITY_CATEGORY,
        ]
    )


def title(event):
    resource_id = event.get("resourceId", "<UNKNOWN_RESOURCE>")

    return f"Azure Advisor Security Recommendation Available for [{resource_id}]"


def alert_context(event):
    context = azure_activity_alert_context(event)
    context["recommendation_name"] = event.deep_get(
        "properties", "recommendationName", default=None
    )
    context["recommendation_impact"] = event.deep_get(
        "properties", "recommendationImpact", default=None
    )
    context["recommendation_category"] = event.deep_get(
        "properties", "recommendationCategory", default=None
    )
    context["recommendation_type"] = event.deep_get(
        "properties", "recommendationType", default=None
    )
    context["recommendation_link"] = event.deep_get(
        "properties", "recommendationResourceLink", default=None
    )
    context["result_description"] = event.get("resultDescription", None)

    return context

AnalysisType: rule
Filename: azure_advisor_security_recommendation.py
RuleID: "Azure.MonitorActivity.Advisor.SecurityRecommendation"
DisplayName: "Azure Advisor Security Recommendation Available"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: Info
Status: Experimental
Description: >
  Detects when Azure Advisor generates a new security recommendation for a resource.
  Azure Advisor analyzes your resource configurations and usage telemetry to recommend solutions
  that can help improve security, cost effectiveness, performance, reliability, and operational excellence.
Runbook: |
  1. Query Azure Monitor Activity logs for all operations on the resourceId in the 48 hours before the alert to identify recent configuration changes
  2. Check if this recommendationType has been triggered for other resources in the tenantId in the past 30 days to identify patterns
  3. Review the recommendationImpact and recommendationName to assess priority and determine if immediate remediation is required
Reference: https://learn.microsoft.com/en-us/azure/advisor/advisor-overview
SummaryAttributes:
  - resourceId
  - properties.recommendationName
  - properties.recommendationImpact

from panther_azureactivity_helpers import (
    azure_activity_alert_context,
    azure_activity_success,
    extract_resource_name_from_id,
)

ALERT_RULES_DELETE = "MICROSOFT.INSIGHTS/ALERTRULES/DELETE"
METRIC_ALERTS_DELETE = "MICROSOFT.INSIGHTS/METRICALERTS/DELETE"


def rule(event):
    operation = event.get("operationName", "").upper()
    return operation in [ALERT_RULES_DELETE, METRIC_ALERTS_DELETE] and azure_activity_success(event)


def title(event):
    alert_rule = event.get("resourceId", "<UNKNOWN_ALERT>")
    return f"Azure Alert Rule deleted [{alert_rule}]"


def alert_context(event):
    context = azure_activity_alert_context(event)

    resource_id = event.get("resourceId", "")
    operation = event.get("operationName", "").upper()

    # Determine resource type based on operation
    if operation == METRIC_ALERTS_DELETE:
        alert_name = extract_resource_name_from_id(resource_id, "metricalerts", default="")
        if alert_name:
            context["alert_rule_name"] = alert_name
            context["alert_type"] = "metric"
    else:
        alert_name = extract_resource_name_from_id(resource_id, "alertrules", default="")
        if alert_name:
            context["alert_rule_name"] = alert_name
            context["alert_type"] = "classic"

    resource_group = extract_resource_name_from_id(resource_id, "resourceGroups", default="")
    if resource_group:
        context["resource_group"] = resource_group

    return context

AnalysisType: rule
Filename: azure_alert_rules_deleted.py
RuleID: "Azure.MonitorActivity.AlertRules.Deleted"
DisplayName: "Azure Alert Rules Deleted"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: Medium
Description: >
  Detects when Azure alert rules are deleted. Deleting alert rules disables security notifications and is a common defense evasion technique.
Reports:
  MITRE ATT&CK:
    - TA0005:T1562.008 # Defense Evasion: Impair Defenses - Disable Cloud Logs
Tags:
  - Defense Evasion
  - Impair Defenses
  - Disable Cloud Logs
Runbook: |
  1. Query Azure Monitor Activity logs for all alert rule and notification operations by the callerIpAddress in the 24 hours before and after this alert to identify if multiple alerts are being deleted
  2. Check if the source IP is associated with known cloud providers, VPN services, or corporate network ranges using threat intelligence
  3. Search for other Azure monitoring or security configuration changes from the same user or IP in the past 7 days to assess if this is part of a defense evasion campaign
Reference: https://learn.microsoft.com/en-us/azure/azure-monitor/alerts/alerts-manage-alert-rules
SummaryAttributes:
  - resourceId
  - callerIpAddress
  - correlationId

from panther_azureactivity_helpers import (
    azure_activity_alert_context,
    azure_activity_success,
    extract_resource_name_from_id,
)

ALERT_SUPPRESSION_WRITE_OPERATION = "MICROSOFT.SECURITY/ALERTSSUPPRESSIONRULES/WRITE"


def rule(event):
    return all(
        [
            event.get("operationName", "").upper() == ALERT_SUPPRESSION_WRITE_OPERATION,
            azure_activity_success(event),
        ]
    )


def title(event):
    resource_id = event.get("resourceId", "<UNKNOWN_RESOURCE>")

    rule_name = extract_resource_name_from_id(
        resource_id, "alertsSuppressionRules", default="<UNKNOWN_RULE_NAME>"
    )

    return f"Azure Alert Suppression Rule Created or Modified: [{rule_name}]"


def alert_context(event):
    context = azure_activity_alert_context(event)

    resource_id = event.get("resourceId", "")
    rule_name = extract_resource_name_from_id(resource_id, "alertsSuppressionRules", default="")
    if rule_name:
        context["suppression_rule_name"] = rule_name

    return context

AnalysisType: rule
Filename: azure_alert_suppression_rule_created.py
RuleID: "Azure.MonitorActivity.Security.AlertSuppressionRuleCreated"
DisplayName: "Azure Alert Suppression Rule Created or Modified"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: Low
Description: >
  Detects when an Azure Security Center alert suppression rule is created or modified.
  Alert suppression rules allow filtering of specific security alerts to reduce noise,
  but adversaries may abuse this feature to silence alerts related to their malicious
  activities. While legitimate use cases exist (suppressing known false positives), new
  suppression rules should be reviewed to ensure they don't hide indicators of compromise.
Reports:
  MITRE ATT&CK:
    - TA0005:T1562 # Defense Evasion: Impair Defenses
Tags:
  - Defense Evasion
  - Impair Defenses
Runbook: |
  1. Query Azure Monitor Activity logs for all security control operations (alert suppression rules, alert rule deletions, diagnostic settings deletions) by the callerIpAddress in the 24 hours before and after the alert
  2. Find all alert suppression rule creations in the past 6 hours to determine if multiple alerts are being suppressed to hide malicious activity
  3. Check if the callerIpAddress has created alert suppression rules in the past 90 days to establish if this is typical security operations activity
Reference: https://learn.microsoft.com/en-us/azure/defender-for-cloud/alerts-suppression-rules
SummaryAttributes:
  - resourceId
  - callerIpAddress
  - correlationId

def rule(event):
    operation_name = event.get("operationName", "")

    if "authentication methods policy update" not in operation_name.lower():
        return False

    old_values = event.deep_walk(
        "properties", "targetResources", "modifiedProperties", "oldValue", default=[]
    )
    new_values = event.deep_walk(
        "properties", "targetResources", "modifiedProperties", "newValue", default=[]
    )

    # Ensure we have lists
    if not isinstance(old_values, list):
        old_values = [old_values] if old_values else []
    if not isinstance(new_values, list):
        new_values = [new_values] if new_values else []

    if len(old_values) != len(new_values):
        # Lists have different lengths; check all values to be safe
        for value in old_values + new_values:
            if isinstance(value, str) and "discoveryUrl" in value:
                return True
        return False

    for old_value, new_value in zip(old_values, new_values):
        if (
            isinstance(old_value, str)
            and isinstance(new_value, str)
            and "discoveryUrl" in old_value
            and "discoveryUrl" in new_value
        ):
            if old_value != new_value:
                return True

    return False


def title(event):
    actor = event.deep_get(
        "properties", "initiatedBy", "user", "userPrincipalName", default="<UNKNOWN_ACTOR>"
    )
    return f"Authentication Methods Policy OIDC Discovery URL Changed by [{actor}]"


def alert_context(event):
    context = {}

    context["operation_name"] = event.get("operationName", "<NO_OPERATION>")
    context["activity_display_name"] = event.deep_get(
        "properties", "activityDisplayName", default="<NO_ACTIVITY>"
    )
    context["category"] = event.get("category", "<NO_CATEGORY>")

    context["initiator_user_id"] = event.deep_get(
        "properties", "initiatedBy", "user", "id", default="<NO_USER_ID>"
    )
    context["initiator_display_name"] = event.deep_get(
        "properties", "initiatedBy", "user", "displayName", default="<NO_DISPLAY_NAME>"
    )
    context["initiator_ip"] = event.deep_get(
        "properties", "initiatedBy", "user", "ipAddress", default="<NO_IP>"
    )

    # Extract OIDC discovery URL changes
    old_values = event.deep_walk(
        "properties", "targetResources", "modifiedProperties", "oldValue", default=[]
    )
    new_values = event.deep_walk(
        "properties", "targetResources", "modifiedProperties", "newValue", default=[]
    )

    if not isinstance(old_values, list):
        old_values = [old_values] if old_values else []
    if not isinstance(new_values, list):
        new_values = [new_values] if new_values else []

    for old_value, new_value in zip(old_values, new_values):
        if (isinstance(old_value, str) and "discoveryUrl" in old_value) or (
            isinstance(new_value, str) and "discoveryUrl" in new_value
        ):
            context["old_discovery_url"] = old_value
            context["new_discovery_url"] = new_value
            break

    return context

AnalysisType: rule
Filename: azure_auth_methods_policy_oidc_change.py
RuleID: "Azure.Audit.OIDC.Changed"
DisplayName: "Azure Authentication Methods Policy OIDC Discovery URL Changed"
Enabled: true
LogTypes:
  - Azure.Audit
Severity: High
Description: >
  Detects modifications to the OIDC discovery URL in Azure Entra ID's Authentication Methods Policy. This technique enables attackers to federate the tenant with attacker-controlled identity providers, bypassing multi-factor authentication and enabling unauthorized access through bring-your-own IdP methods.
Tags:
  - Persistence
  - Modify Authentication Process
Reports:
  MITRE ATT&CK:
    - TA0003:T1556
    - TA0003:T1556.009
Runbook: |
  1. Query Azure.Audit logs for all authentication policy changes by properties:initiatedBy:user:userPrincipalName in the 7 days before and after this change to determine if this modification was part of a broader attack campaign
  2. Immediately verify the new OIDC discovery URL with your identity team to confirm it points to a legitimate identity provider owned by your organization and not an attacker-controlled domain
  3. Query Azure.Audit and sign-in logs for all authentication events in the 24 hours after the policy change to identify any suspicious token issuance or unauthorized access attempts, and if the modification was unauthorized, immediately revert the authentication methods policy, revoke all active sessions, and review all API access and resource modifications that occurred during the compromise window
Reference: https://github.com/elastic/detection-rules/blob/main/rules/integrations/azure/persistence_entra_id_oidc_discovery_url_change.toml
SummaryAttributes:
  - properties:initiatedBy:user:userPrincipalName
  - properties:targetResources:displayName
  - properties:initiatedBy:user:ipAddress

from panther_azureactivity_helpers import (
    azure_activity_alert_context,
    azure_activity_success,
    extract_resource_name_from_id,
)

AUTOMATION_ACCOUNT_WRITE = "MICROSOFT.AUTOMATION/AUTOMATIONACCOUNTS/WRITE"


def rule(event):
    return event.get(
        "operationName", ""
    ).upper() == AUTOMATION_ACCOUNT_WRITE and azure_activity_success(event)


def title(event):
    resource_id = event.get("resourceId", "<UNKNOWN_RESOURCE>")

    account_name = extract_resource_name_from_id(
        resource_id, "automationAccounts", default="<UNKNOWN_ACCOUNT>"
    )

    return f"Azure Automation Account Created: [{account_name}]"


def alert_context(event):
    context = azure_activity_alert_context(event)

    resource_id = event.get("resourceId", "")
    account_name = extract_resource_name_from_id(resource_id, "automationAccounts", default="")
    if account_name:
        context["automation_account_name"] = account_name

    return context

AnalysisType: rule
Filename: azure_automation_account_created.py
RuleID: "Azure.MonitorActivity.Automation.AccountCreated"
DisplayName: "Azure Automation Account Created"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: Info
Description: >
  Detects when an Azure Automation account is created. Azure Automation accounts can be used to
  automate management tasks and orchestrate actions across systems. Adversaries may create
  Automation accounts to maintain persistence in their target's environment by leveraging managed
  identities and runbooks to execute code with elevated privileges.
Reports:
  MITRE ATT&CK:
    - TA0003:T1078 # Persistence: Valid Accounts
    - TA0005:T1078 # Defense Evasion: Valid Accounts
Tags:
  - Persistence
  - Defense Evasion
  - Valid Accounts
Runbook: |
  1. Query Azure Monitor Activity logs for all automation-related operations (runbook creation, webhook creation, job executions) for the newly created automation account in the 6 hours after the account was created
  2. Find all automation account creations by the callerIpAddress in the past 24 hours to identify if multiple accounts are being created
  3. Check if the callerIpAddress has created automation accounts in the past 90 days to establish if this is typical infrastructure deployment behavior
Reference: https://github.com/elastic/detection-rules/blob/main/rules/integrations/azure/persistence_automation_account_created.toml
SummaryAttributes:
  - resourceId
  - callerIpAddress
  - correlationId

from panther_azureactivity_helpers import (
    azure_activity_alert_context,
    azure_activity_success,
    extract_resource_name_from_id,
)

RUNBOOK_OPERATIONS = [
    "MICROSOFT.AUTOMATION/AUTOMATIONACCOUNTS/RUNBOOKS/DRAFT/WRITE",
    "MICROSOFT.AUTOMATION/AUTOMATIONACCOUNTS/RUNBOOKS/WRITE",
    "MICROSOFT.AUTOMATION/AUTOMATIONACCOUNTS/RUNBOOKS/PUBLISH/ACTION",
]


def rule(event):
    return event.get("operationName", "").upper() in RUNBOOK_OPERATIONS and azure_activity_success(
        event
    )


def title(event):
    resource_id = event.get("resourceId", "<UNKNOWN_RESOURCE_ID>")

    runbook_name = extract_resource_name_from_id(
        resource_id, "runbooks", default="<UNKNOWN_RUNBOOK_NAME>"
    )

    operation = event.get("operationName", "").upper()
    action = "Modified"
    if "PUBLISH" in operation:
        action = "Published"
    elif "DRAFT" in operation:
        action = "Draft Created"

    return f"Azure Automation Runbook {action}: [{runbook_name}]"


def alert_context(event):
    context = azure_activity_alert_context(event)

    resource_id = event.get("resourceId", "")

    runbook_name = extract_resource_name_from_id(resource_id, "runbooks", default="")
    if runbook_name:
        context["runbook_name"] = runbook_name

    automation_account_name = extract_resource_name_from_id(
        resource_id, "automationAccounts", default=""
    )
    if automation_account_name:
        context["automation_account_name"] = automation_account_name

    return context

AnalysisType: rule
Filename: azure_automation_runbook_created_or_modified.py
RuleID: "Azure.MonitorActivity.Automation.RunbookCreatedOrModified"
DisplayName: "Azure Automation Runbook Created or Modified"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: Info
Description: >
  Detects when an Azure Automation runbook is created, modified, or published. Runbooks contain
  executable code that can automate tasks within Azure environments. Adversaries may abuse runbooks
  for persistence, privilege escalation, or execution of malicious scripts. This rule monitors draft
  creation, runbook modifications, and publishing activities that could indicate unauthorized automation
  code deployment.
Reports:
  MITRE ATT&CK:
    - TA0003:T1078.004 # Persistence: Cloud Accounts
    - TA0002:T1059 # Execution: Command and Scripting Interpreter
Tags:
  - AZT302
  - AZT302.1
  - AZT302.2
  - AZT302.3
  - AZT601
  - AZT601.5
  - AZT602
  - AZT605.2
  - Persistence
  - Execution
  -  Cloud Accounts
  - Command and Scripting Interpreter
Runbook: |
  1. Query Azure Monitor Activity logs for all operations by the callerIpAddress in the 24 hours before and after the alert to establish activity patterns
  2. Check if the caller has created or modified automation runbooks in the past 90 days to determine if this is typical behavior
  3. Search for other automation-related activities (webhook creation, automation account creation, runbook job executions) from the same caller in the 6 hours around the alert
Reference: https://github.com/elastic/detection-rules/blob/main/rules/integrations/azure/execution_azure_automation_runbook_created_or_modified.toml
SummaryAttributes:
  - resourceId
  - callerIpAddress
  - correlationId

from panther_azureactivity_helpers import (
    azure_activity_alert_context,
    azure_activity_success,
    extract_resource_name_from_id,
)

RUNBOOK_DELETE_OPERATION = "MICROSOFT.AUTOMATION/AUTOMATIONACCOUNTS/RUNBOOKS/DELETE"


def rule(event):
    return event.get(
        "operationName", ""
    ).upper() == RUNBOOK_DELETE_OPERATION and azure_activity_success(event)


def title(event):
    resource_id = event.get("resourceId", "<UNKNOWN_RESOURCE>")

    runbook_name = extract_resource_name_from_id(
        resource_id, "runbooks", default="<UNKNOWN_RUNBOOK_NAME>"
    )

    return f"Azure Automation Runbook Deleted: [{runbook_name}]"


def alert_context(event):
    context = azure_activity_alert_context(event)

    resource_id = event.get("resourceId", "")

    runbook_name = extract_resource_name_from_id(resource_id, "runbooks", default="")
    if runbook_name:
        context["runbook_name"] = runbook_name

    automation_account_name = extract_resource_name_from_id(
        resource_id, "automationAccounts", default=""
    )
    if automation_account_name:
        context["automation_account_name"] = automation_account_name

    return context

AnalysisType: rule
Filename: azure_automation_runbook_deleted.py
RuleID: "Azure.MonitorActivity.Automation.RunbookDeleted"
DisplayName: "Azure Automation Runbook Deleted"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: Info
Description: >
  Detects when an Azure Automation runbook is deleted. Adversaries may delete runbooks to cover
  their tracks after using them for malicious purposes, to disrupt automated security responses,
  or to eliminate forensic evidence. Legitimate runbook deletions should be rare and controlled
  through change management processes.
Reports:
  MITRE ATT&CK:
    - TA0005:T1070 # Defense Evasion: Indicator Removal
Tags:
  - Defense Evasion
  - Indicator Removal
Runbook: |
  1. Query Azure Monitor Activity logs for all runbook operations (create, modify, delete) by the callerIpAddress in the 24 hours before the deletion to identify if the runbook was recently created by the same actor
  2. Find all runbook deletions in the past 6 hours to determine if this is part of a larger cleanup operation
  3. Search for other defense evasion activities (diagnostic settings deletions, alert rule deletions, event hub deletions) from the same caller in the 24 hours around the alert
Reference: https://github.com/elastic/detection-rules/blob/main/rules/integrations/azure/defense_evasion_azure_automation_runbook_deleted.toml
SummaryAttributes:
  - resourceId
  - callerIpAddress
  - correlationId

from panther_azureactivity_helpers import (
    azure_activity_alert_context,
    azure_activity_success,
    extract_resource_name_from_id,
)

SCHEDULE_OPERATIONS = [
    "MICROSOFT.AUTOMATION/AUTOMATIONACCOUNTS/SCHEDULES/WRITE",
]


def rule(event):
    return event.get("operationName", "").upper() in SCHEDULE_OPERATIONS and azure_activity_success(
        event
    )


def title(event):
    resource_id = event.get("resourceId", "<UNKNOWN_RESOURCE>")
    caller = event.get("callerIpAddress", "<UNKNOWN_CALLER>")

    schedule_name = extract_resource_name_from_id(
        resource_id, "schedules", default="<UNKNOWN_SCHEDULE_NAME>"
    )

    return f"Azure Automation Schedule Created or Modified: [{schedule_name}] by [{caller}]"


def alert_context(event):
    context = azure_activity_alert_context(event)

    resource_id = event.get("resourceId", "")

    schedule_name = extract_resource_name_from_id(resource_id, "schedules", default="")
    if schedule_name:
        context["schedule_name"] = schedule_name

    automation_account_name = extract_resource_name_from_id(
        resource_id, "automationAccounts", default=""
    )
    if automation_account_name:
        context["automation_account_name"] = automation_account_name

    return context

AnalysisType: rule
Filename: azure_automation_schedule_created.py
RuleID: "Azure.MonitorActivity.Automation.ScheduleCreated"
DisplayName: "Azure Automation Schedule Created or Modified"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: Info
Description: >
  Detects when an Azure Automation schedule is created or modified. Schedules define when and how often
  automation runbooks execute. Adversaries may create or modify schedules to establish persistence by
  executing malicious runbooks at regular intervals or specific times. This technique allows attackers
  to maintain access and execute commands without requiring direct interaction.
Reports:
  MITRE ATT&CK:
    - TA0003:T1053.005 # Persistence: Scheduled Task/Job - Scheduled Task
    - TA0004:T1068 # Privilege Escalation: Exploitation for Privilege Escalation
Tags:
  - AZT505
  - Persistence
  - Privilege Escalation
  - Scheduled Task
  - Exploitation for Privilege Escalation
Runbook: |
  1. Find all automation account operations by the callerIpAddress in the 24 hours before and after this alert to identify if runbooks were created or modified alongside this schedule
  2. Query for runbook job execution events from the automation account in the 6 hours after the schedule creation to determine if the schedule has already triggered
  3. Check if the callerIpAddress is associated with known cloud providers, VPNs, or corporate network ranges used by authorized DevOps personnel
Reference: https://microsoft.github.io/Azure-Threat-Research-Matrix/Persistence/AZT505/AZT505-1
SummaryAttributes:
  - resourceId
  - callerIpAddress
  - correlationId

from panther_azureactivity_helpers import (
    azure_activity_alert_context,
    azure_activity_success,
    extract_resource_name_from_id,
)

AUTOMATION_WEBHOOK_OPERATIONS = [
    "MICROSOFT.AUTOMATION/AUTOMATIONACCOUNTS/WEBHOOKS/ACTION",
    "MICROSOFT.AUTOMATION/AUTOMATIONACCOUNTS/WEBHOOKS/WRITE",
]


def rule(event):
    return event.get(
        "operationName", ""
    ).upper() in AUTOMATION_WEBHOOK_OPERATIONS and azure_activity_success(event)


def title(event):
    resource_id = event.get("resourceId", "<UNKNOWN_RESOURCE>")

    webhook_name = extract_resource_name_from_id(
        resource_id, "webhooks", default="<UNKNOWN_WEBHOOK_NAME>"
    )

    return f"Azure Automation Webhook Created: [{webhook_name}]"


def alert_context(event):
    context = azure_activity_alert_context(event)

    resource_id = event.get("resourceId", "")

    webhook_name = extract_resource_name_from_id(resource_id, "webhooks", default="")
    if webhook_name:
        context["webhook_name"] = webhook_name

    automation_account_name = extract_resource_name_from_id(
        resource_id, "automationAccounts", default=""
    )
    if automation_account_name:
        context["automation_account_name"] = automation_account_name

    return context

AnalysisType: rule
Filename: azure_automation_webhook_created.py
RuleID: "Azure.MonitorActivity.Automation.WebhookCreated"
DisplayName: "Azure Automation Webhook Created"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: Info
Description: >
  Detects when an Azure Automation webhook is created. A webhook uses a custom URL passed to
  Azure Automation along with a data payload specific to the runbook. Adversaries may exploit
  this capability to trigger runbooks containing malicious code for persistence or to execute
  unauthorized actions in the environment.
Reports:
  MITRE ATT&CK:
    - TA0003:T1546 # Persistence: Event Triggered Execution
    - TA0042:T1608 # Resource Development: Stage Capabilities
Tags:
- AZT502
- AZT503
- AZT503.3
- Persistence
- Resource Development
- Event Triggered Execution
- Stage Capabilities
Runbook: |
  1. Query Azure Monitor Activity logs for all automation account operations (account creation, runbook creation, webhook creation) by the callerIpAddress in the 24 hours before and after the alert to identify a sequence of persistence activities
  2. Find all webhook creations in the past 6 hours to determine if multiple persistence mechanisms are being established
  3. Check if the callerIpAddress has created webhooks in the past 90 days to determine if this is typical automation workflow deployment
Reference: https://github.com/elastic/detection-rules/blob/main/rules/integrations/azure/persistence_automation_webhook_created.toml
SummaryAttributes:
  - resourceId
  - callerIpAddress
  - correlationId

from panther_azuresignin_helpers import (
    azure_signin_alert_context,
    azure_signin_success,
    is_sign_in_event,
)

# Microsoft Broker Client Application ID
BROKER_CLIENT_APP_ID = "29d9ed98-a469-4536-ade2-f981bc1d605e"


def rule(event):
    if not is_sign_in_event(event) or not azure_signin_success(event):
        return False

    auth_protocol = event.deep_get("properties", "authenticationProtocol", default="").lower()
    if auth_protocol != "devicecode":
        return False

    app_id = event.deep_walk(
        "properties", "conditionalAccessAudiences", "applicationId", default=""
    )

    # deep_walk returns a list if multiple values found, or a string if one value
    if isinstance(app_id, list):
        return BROKER_CLIENT_APP_ID in app_id
    return BROKER_CLIENT_APP_ID == app_id


def title(event):
    user_principal_name = event.deep_get(
        "properties", "userPrincipalName", default="<UNKNOWN_USER>"
    )
    source_ip = event.deep_get("properties", "ipAddress", default="<UNKNOWN_IP>")

    return (
        f"Device Code Authentication with Broker Client: User [{user_principal_name}] "
        f"from IP [{source_ip}]"
    )


def alert_context(event):
    context = azure_signin_alert_context(event)
    return context

AnalysisType: rule
Filename: azure_device_code_broker_client.py
RuleID: "Azure.Audit.DeviceCodeBrokerClient"
DisplayName: "Azure Device Code Authentication with Broker Client"
Enabled: true
LogTypes:
  - Azure.Audit
Severity: Medium
DedupPeriodMinutes: 60
Description: >
  Detects device code authentication using the Microsoft Broker Client application, which may indicate
  Primary Refresh Token (PRT) abuse. Device code flow allows adversaries to trick users into entering
  codes on attacker-controlled applications. When combined with Microsoft Broker Client (app ID
  29d9ed98-a469-4536-ade2-f981bc1d605e), this may indicate PRT theft or replay attacks that bypass MFA
  and Conditional Access policies.
Tags:
  - Initial Access
  - Phishing
  - Use Alternate Authentication Material
  - Valid Accounts
Reports:
  MITRE ATT&CK:
    - TA0001:T1566
    - TA0001:T1566.002
    - TA0005:T1550
    - TA0005:T1550.001
    - TA0008:T1078
    - TA0008:T1078.004
Runbook: |
  1. Query Azure.Audit logs for all device code authentication attempts by properties:userPrincipalName in the 24 hours before and after this event to identify the scope of potential phishing campaign or PRT abuse affecting this user
  2. Check properties:deviceDetail to verify if the device is registered and compliant in Azure AD, and review whether the user recognizes this device and the sign-in activity as legitimate
  3. Search for other users in the organization with similar device code authentication patterns using the broker client application ID to determine if this is an isolated incident or part of a broader attack campaign targeting multiple accounts
Reference: https://github.com/elastic/detection-rules/blob/main/rules/integrations/azure/initial_access_entra_id_device_code_auth_with_broker_client.toml
SummaryAttributes:
  - properties:userPrincipalName
  - callerIpAddress
  - properties:deviceDetail:deviceId
  - properties:appDisplayName

from panther_azureactivity_helpers import (
    azure_activity_alert_context,
    azure_activity_success,
    extract_resource_name_from_id,
)

DIAGNOSTIC_SETTINGS_DELETE = "MICROSOFT.INSIGHTS/DIAGNOSTICSETTINGS/DELETE"


def rule(event):
    return event.get(
        "operationName", ""
    ).upper() == DIAGNOSTIC_SETTINGS_DELETE and azure_activity_success(event)


def title(event):
    resource_id = event.get("resourceId", "")
    resource = extract_resource_name_from_id(
        resource_id, "diagnosticSettings", default="<UNKNOWN_RESOURCE>"
    )

    return f"Azure Diagnostic Settings deleted on [{resource}]"


def alert_context(event):
    context = azure_activity_alert_context(event)

    resource_id = event.get("resourceId", "")

    diagnostic_setting_name = extract_resource_name_from_id(
        resource_id, "diagnosticSettings", default=""
    )
    if diagnostic_setting_name:
        context["diagnostic_setting_name"] = diagnostic_setting_name

    return context

AnalysisType: rule
Filename: azure_diagnostic_settings_deleted.py
RuleID: "Azure.MonitorActivity.DiagnosticSettings.Deleted"
DisplayName: "Azure Diagnostic Settings Deleted"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: Medium
Description: >
  Detects when Azure diagnostic settings are deleted. Deleting diagnostic settings disables logging and monitoring, which is a common technique to hide malicious activity.
Reports:
  MITRE ATT&CK:
    - TA0005:T1562.008 # Defense Evasion: Impair Defenses - Disable Cloud Logs
Tags:
  - Defense Evasion
  - Impair Defenses
  - Disable Cloud Logs
Runbook: |
  1. Query Azure Monitor Activity logs for all diagnostic settings and monitoring operations by the callerIpAddress in the 24 hours before and after this alert to identify if multiple resources are being de-monitored
  2. Check if the source IP is associated with known cloud providers, VPN services, or corporate network ranges using threat intelligence
  3. Search for other Azure resource modifications or suspicious activities from the same user or IP in the past 7 days to assess if this is part of a defense evasion campaign
Reference: https://docs.datadoghq.com/security/default_rules/apn-0ib-a6f/
SummaryAttributes:
  - resourceId
  - callerIpAddress
  - correlationId

from panther_azureactivity_helpers import (
    azure_activity_alert_context,
    azure_activity_success,
    extract_resource_name_from_id,
)

DISK_DELETE = "MICROSOFT.COMPUTE/DISKS/DELETE"


def rule(event):
    return event.get("operationName", "").upper() == DISK_DELETE and azure_activity_success(event)


def title(event):
    resource_id = event.get("resourceId", "")
    disk = extract_resource_name_from_id(resource_id, "disks", default="<UNKNOWN_DISK>")

    return f"Azure disk deleted [{disk}]"


def alert_context(event):
    context = azure_activity_alert_context(event)

    resource_id = event.get("resourceId", "")

    disk_name = extract_resource_name_from_id(resource_id, "disks", default="")
    if disk_name:
        context["disk_name"] = disk_name

    resource_group = extract_resource_name_from_id(resource_id, "resourceGroups", default="")
    if resource_group:
        context["resource_group"] = resource_group

    return context

AnalysisType: rule
Filename: azure_disk_deleted.py
RuleID: "Azure.MonitorActivity.Compute.DiskDeleted"
DisplayName: "Azure Disk Deleted"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: Info
Status: Experimental
Description: >
  Detects when an Azure managed disk is deleted.
  Unauthorized disk deletion can indicate ransomware activity where attackers destroy data or delete backup disks to prevent recovery.
  This may also indicate legitimate cleanup operations.
Reports:
  MITRE ATT&CK:
    - TA0040:T1485 # Impact: Data Destruction
    - TA0040:T1490 # Impact: Inhibit System Recovery
Tags:
  - Impact
  - Data Destruction
  - Inhibit System Recovery
  - Ransomware
Runbook: |
  1. Query Azure Monitor Activity logs for all disk operations by the callerIpAddress in the 24 hours before and after the alert to identify if multiple disks are being deleted in sequence
  2. Find all compute resource deletion events from the same caller in the past 7 days to assess if this is part of a broader data destruction pattern
  3. Check if the source IP matches known VPN ranges or corporate network addresses associated with authorized administrators
Reference: https://learn.microsoft.com/en-us/rest/api/compute/disks/delete?view=rest-compute-2025-04-01#:~:text=Deletes%20a%20disk.,version=2025-01-02
SummaryAttributes:
  - resourceId
  - callerIpAddress
  - correlationId

def rule(event):
    operation_name = event.get("operationName", "")

    # Branch 1: For "Set federation settings on domain" any change is suspicious
    if "set federation settings on domain" in operation_name.lower():
        return True

    # Branch 2: For "Set domain authentication" check if LiveType property changed to Federated
    if "set domain authentication" in operation_name.lower():
        display_names = event.deep_walk(
            "properties", "targetResources", "modifiedProperties", "displayName", default=[]
        )
        new_values = event.deep_walk(
            "properties", "targetResources", "modifiedProperties", "newValue", default=[]
        )

        # Ensure we have lists (deep_walk returns single value if only one result)
        if not isinstance(display_names, list):
            display_names = [display_names] if display_names else []
        if not isinstance(new_values, list):
            new_values = [new_values] if new_values else []

        if len(display_names) != len(new_values):
            # Lists have different lengths; check all values with consistent approach
            if "LiveType" in display_names and any(
                isinstance(val, str) and "Federated" in val for val in new_values
            ):
                return True
            return False

        # Check if the same property has displayName="LiveType" AND newValue contains "Federated"
        for display_name, new_value in zip(display_names, new_values):
            if (
                display_name == "LiveType"
                and isinstance(new_value, str)
                and "Federated" in new_value
            ):
                return True

    return False


def title(event):
    actor = event.deep_get(
        "properties", "initiatedBy", "user", "userPrincipalName", default="<UNKNOWN_ACTOR>"
    )

    return f"Domain Federation Trust Settings Modified by [{actor}] "


def alert_context(event):
    context = {}

    # Add federation-specific context
    context["operation_name"] = event.get("operationName", "<NO_OPERATION>")
    context["activity_display_name"] = event.deep_get(
        "properties", "activityDisplayName", default="<NO_ACTIVITY>"
    )
    context["category"] = event.get("category", "<NO_CATEGORY>")

    # Add initiator details
    context["initiator_user_id"] = event.deep_get(
        "properties", "initiatedBy", "user", "id", default="<NO_USER_ID>"
    )
    context["initiator_display_name"] = event.deep_get(
        "properties", "initiatedBy", "user", "displayName", default="<NO_DISPLAY_NAME>"
    )
    context["initiator_ip"] = event.deep_get(
        "properties", "initiatedBy", "user", "ipAddress", default="<NO_IP>"
    )

    return context

AnalysisType: rule
Filename: azure_domain_trust_settings_modified.py
RuleID: "Azure.Audit.DomainSettingsModified"
DisplayName: "Azure Domain Federation Settings Modified"
Enabled: true
LogTypes:
  - Azure.Audit
Severity: Medium
Description: >
  Detects modifications to domain federation settings in Microsoft Entra ID, including changes to
  federation trust configurations and OIDC discovery endpoints. Adversaries who compromise administrative
  accounts may modify these settings to federate the tenant with attacker-controlled identity providers,
  enabling unauthorized access and MFA bypass. This technique allows attackers to establish persistent
  access by redirecting authentication to malicious infrastructure.
Tags:
  - Persistence
  - Modify Authentication Process
Reports:
  MITRE ATT&CK:
    - TA0003:T1556
    - TA0003:T1556.006
Runbook: |
  1. Query Azure.Audit logs for all federation-related operations by properties:initiatedBy:user:userPrincipalName in the 7 days before and after this change to determine if this modification was part of a broader attack campaign involving administrative credential compromise or privilege escalation
  2. Verify with your identity team and the initiating administrator whether this federation settings change was authorized through proper change management procedures, and review the new OIDC discovery endpoint URL to confirm it points to a legitimate identity provider owned by your organization and not an attacker-controlled domain
  3. Query Azure.Audit logs and sign-in logs for all authentication events using the affected domain in the 24 hours after the federation change to identify any suspicious token issuance or unauthorized access attempts, and if the modification was unauthorized, immediately revert the federation settings, revoke all active sessions for affected users, and review all API access and resource modifications that occurred during the compromise window
Reference: https://github.com/Azure/Azure-Sentinel/blob/master/Solutions/Microsoft%20Entra%20ID/Analytic%20Rules/ADFSDomainTrustMods.yaml
SummaryAttributes:
  - properties:initiatedBy:user:userPrincipalName
  - properties:targetResources:displayName
  - properties:initiatedBy:user:ipAddress

from panther_azureactivity_helpers import (
    azure_activity_alert_context,
    azure_activity_success,
    extract_resource_name_from_id,
)

EVENT_HUB_DELETE_OPERATION = "MICROSOFT.EVENTHUB/NAMESPACES/EVENTHUBS/DELETE"


def rule(event):
    return all(
        [
            event.get("operationName", "").upper() == EVENT_HUB_DELETE_OPERATION,
            azure_activity_success(event),
        ]
    )


def title(event):
    resource_id = event.get("resourceId", "")
    eventhub = extract_resource_name_from_id(resource_id, "eventhubs", default="<UNKNOWN_EVENTHUB>")
    return f"Azure Event Hub Deleted: [{eventhub}]"


def alert_context(event):
    context = azure_activity_alert_context(event)

    resource_id = event.get("resourceId", "")

    eventhub_name = extract_resource_name_from_id(resource_id, "eventhubs", default="")
    if eventhub_name:
        context["eventhub_name"] = eventhub_name

    namespace_name = extract_resource_name_from_id(resource_id, "namespaces", default="")
    if namespace_name:
        context["namespace_name"] = namespace_name

    resource_group = extract_resource_name_from_id(resource_id, "resourceGroups", default="")
    if resource_group:
        context["resource_group"] = resource_group

    return context

AnalysisType: rule
Filename: azure_event_hub_deleted.py
RuleID: "Azure.MonitorActivity.EventHub.Deleted"
DisplayName: "Azure Event Hub Deleted"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: High
Description: >
  Detects when an Azure Event Hub is deleted. Event Hubs are critical event processing services
  that ingest and process large volumes of data for log collection, SIEM ingestion, and real-time
  analytics. Adversaries may delete Event Hubs to evade detection by disrupting data flows and
  erasing evidence of their malicious activities. Deletion of Event Hubs used for security logging
  can blind security teams to ongoing attacks.
Reports:
  MITRE ATT&CK:
    - TA0005:T1562.008 # Defense Evasion: Impair Defenses - Disable or Modify Cloud Logs
Tags:
  - Defense Evasion
  - Impair Defenses
  - Disable Cloud Logs
Runbook: |
  1. Query Azure Monitor Activity logs for all logging infrastructure operations (event hub deletions, diagnostic settings deletions, log analytics workspace deletions) by the callerIpAddress in the 24 hours before and after the alert
  2. Find all event hub deletions in the past 6 hours to determine if this is part of a coordinated attack on security logging infrastructure
  3. Check if the callerIpAddress has deleted event hubs in the past 90 days to establish if this is typical infrastructure management activity
Reference: https://learn.microsoft.com/en-us/azure/event-hubs/event-hubs-about
SummaryAttributes:
  - resourceId
  - callerIpAddress
  - correlationId

from panther_azuresignin_helpers import azure_signin_alert_context

MICROSOFT_ASNS = {
    "8075",  # Microsoft Corporation
    "8068",  # Microsoft Corporation
    "8069",  # Microsoft Corporation
    "8070",  # Microsoft Corporation
    "12076",  # Microsoft Azure
}


def rule(event):
    error_code = event.deep_get("properties", "status", "errorCode", default=None)
    asn = event.deep_get("properties", "autonomousSystemNumber", default="")
    user_principal_name = event.deep_get("properties", "userPrincipalName", default="")

    # Exclude Microsoft ASN
    return all([error_code == 50053, asn not in MICROSOFT_ASNS, user_principal_name])


def title(event):
    user_principal_name = event.deep_get(
        "properties", "userPrincipalName", default="<UNKNOWN_USER>"
    )

    return f"Excessive Account Lockouts Detected for [{user_principal_name}]"


def alert_context(event):
    context = azure_signin_alert_context(event)

    # Add lockout-specific context
    context["error_code"] = event.deep_get("properties", "status", "errorCode", default=None)
    context["failure_reason"] = event.deep_get(
        "properties", "status", "failureReason", default="<NO_REASON>"
    )
    context["user_agent"] = event.deep_get("properties", "userAgent", default="<NO_USER_AGENT>")
    context["app_display_name"] = event.deep_get("properties", "appDisplayName", default="<NO_APP>")
    context["location_city"] = event.deep_get("properties", "location", "city", default="<NO_CITY>")
    context["location_country"] = event.deep_get(
        "properties", "location", "countryOrRegion", default="<NO_COUNTRY>"
    )
    context["asn"] = event.deep_get("properties", "autonomousSystemNumber", default="<NO_ASN>")

    return context

AnalysisType: rule
Filename: azure_excessive_account_lockouts.py
RuleID: "Azure.Audit.ExcessiveAccountLockouts"
DisplayName: "Azure Excessive Account Lockouts"
Enabled: true
Status: Experimental
LogTypes:
  - Azure.Audit
Severity: High
DedupPeriodMinutes: 60
Threshold: 20
Description: >
  Detects high volumes of failed Microsoft Entra ID sign-in attempts resulting in account lockouts,
  indicating potential brute-force credential attacks such as password spraying, password guessing, or
  credential stuffing. When adversaries repeatedly attempt authentication with incorrect credentials,
  Entra ID Smart Lockout policies trigger account lockouts (error code 50053).
Tags:
  - Credential Access
  - Brute Force
Reports:
  MITRE ATT&CK:
    - TA0006:T1110
    - TA0006:T1110.001
    - TA0006:T1110.003
    - TA0006:T1110.004
Runbook: |
  1. Query Azure.Audit logs for all authentication attempts from the attacking callerIpAddress in the 60 minutes before and after the alert to identify the full scope of targeted accounts and any successful compromises
  2. Analyze properties:userAgent strings for automation indicators like Python, PowerShell, curl, or wget which suggest scripted attacks rather than manual login attempts
  3. Check if any of the affected user accounts had successful sign-ins from the same callerIpAddress before or after the lockouts, which would indicate successful credential compromise requiring immediate password reset and session revocation
Reference: https://github.com/elastic/detection-rules/blob/main/rules/integrations/azure/credential_access_entra_id_excessive_account_lockouts.toml
SummaryAttributes:
  - callerIpAddress
  - properties:userPrincipalName
  - properties:userAgent
  - properties:location:city
  - properties:location:countryOrRegion

from panther_azureactivity_helpers import azure_activity_alert_context, azure_activity_success

PUBLIC_IP_READ = "MICROSOFT.NETWORK/PUBLICIPADDRESSES/READ"
VM_READ = "MICROSOFT.COMPUTE/VIRTUALMACHINES/READ"


def rule(event):
    operation = event.get("operationName", "").upper()
    return all(
        [
            operation in [PUBLIC_IP_READ, VM_READ],
            azure_activity_success(event),
        ]
    )


def title(event):
    resource_id = event.get("resourceId", "<UNKNOWN_RESOURCE>")
    operation = event.get("operationName", "").upper()

    if operation == PUBLIC_IP_READ:
        resource_type = "Public IP Address"
    elif operation == VM_READ:
        resource_type = "Virtual Machine"
    else:
        resource_type = "Resource"

    return f"Azure Excessive {resource_type} Read on [{resource_id}]"


def alert_context(event):
    return azure_activity_alert_context(event)

AnalysisType: rule
Filename: azure_network_ip_discovery.py
RuleID: "Azure.MonitorActivity.Network.IPDiscovery"
DisplayName: "Azure Excessive IP and VM Discovery"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: Medium
Threshold: 50
Description: >
  Detects excessive read operations on Azure public IP addresses and virtual machines. Adversaries
  may enumerate public IPs and virtual machines to identify external attack surfaces, map network
  topology, and identify potential targets for exploitation. This reconnaissance pattern often
  precedes lateral movement attempts, privilege escalation, or targeted attacks. The threshold-based
  detection triggers when the same resource type is read excessively within a time window.
Reports:
  MITRE ATT&CK:
    - TA0007:T1046 # Discovery: Network Service Discovery
    - TA0007:T1018 # Discovery: Remote System Discovery
    - TA0043:T1595.002 # Reconnaissance: Active Scanning - Vulnerability Scanning
Tags:
  - AZT102
  - Discovery
  - Reconnaissance
  - Network Service Discovery
  - Remote System Discovery
  - Active Scanning
  - Vulnerability Scanning
Runbook: |
  1. Query Azure Monitor Activity logs for all public IP address and virtual machine read operations by the callerIpAddress in the 24 hours before and after the alert to identify the scope of reconnaissance activity
  2. Find all resource read operations (network security groups, route tables, virtual networks, storage accounts) by the same callerIpAddress in the 6 hours before and after the alert to determine if this is part of broader reconnaissance
  3. Check if the callerIpAddress has performed similar high-volume read operations in the past 90 days to establish if this is normal administrative activity or anomalous behavior
Reference: https://microsoft.github.io/Azure-Threat-Research-Matrix/Reconnaissance/AZT101/AZT102/
SummaryAttributes:
  - resourceId
  - location
  - callerIpAddress
  - correlationId

from panther_azureactivity_helpers import azure_activity_alert_context, azure_activity_success

NETWORK_READ = "MICROSOFT.NETWORK/NETWORKSECURITYGROUP/READ"


def rule(event):
    return all(
        [
            event.get("operationName", "").upper() == NETWORK_READ,
            azure_activity_success(event),
        ]
    )


def title(event):
    resource_id = event.get("resourceId", "<UNKNOWN_RESOURCE>")
    return f"Azure Excessive Network Read on [{resource_id}]"


def alert_context(event):
    return azure_activity_alert_context(event)

AnalysisType: rule
Filename: azure_network_port_mapping.py
RuleID: "Azure.MonitorActivity.Network.PortMapping"
DisplayName: "Azure Excessive Network Security Group Read"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: Medium
Threshold: 50
Description: >
  Detects excessive read operations on Azure Network Security Groups. Adversaries may repeatedly
  read Network Security Group configurations to map network ports and firewall rules as part of
  reconnaissance activities. This pattern can indicate an attacker attempting to understand network
  security controls before launching lateral movement or exfiltration attacks. The threshold-based
  detection triggers when the same resource is read excessively within a time window.
Reports:
  MITRE ATT&CK:
    - TA0007:T1046 # Discovery: Network Service Discovery
    - TA0043:T1595.002 # Reconnaissance: Active Scanning - Vulnerability Scanning
Tags:
  - AZT101
  - Discovery
  - Reconnaissance
  - Network Service Discovery
  - Active Scanning
  - Vulnerability Scanning
Runbook: |
  1. Query Azure Monitor Activity logs for all Network Security Group read operations by the callerIpAddress in the 24 hours before and after the alert
  2. Check if the same caller has performed other reconnaissance activities (reading route tables, virtual networks, or other network configurations)
  3. Review the callerIpAddress and caller identity to determine if this is legitimate administrative activity or unauthorized access
  4. Verify if the threshold (50 reads) aligns with normal operational patterns for your environment
Reference: https://microsoft.github.io/Azure-Threat-Research-Matrix/Reconnaissance/AZT101/AZT101/
SummaryAttributes:
  - resourceId
  - location
  - callerIpAddress
  - correlationId

from panther_azureactivity_helpers import (
    azure_activity_alert_context,
    azure_activity_success,
    extract_resource_name_from_id,
)

FIREWALL_POLICY_DELETE_OPERATION = "MICROSOFT.NETWORK/FIREWALLPOLICIES/DELETE"


def rule(event):
    return event.get(
        "operationName", ""
    ).upper() == FIREWALL_POLICY_DELETE_OPERATION and azure_activity_success(event)


def title(event):
    resource_id = event.get("resourceId", "<UNKNOWN_RESOURCE>")

    policy_name = extract_resource_name_from_id(
        resource_id, "firewallPolicies", default="<UNKNOWN_POLICY>"
    )

    return f"Azure Firewall Policy Deleted: [{policy_name}]"


def alert_context(event):
    context = azure_activity_alert_context(event)

    resource_id = event.get("resourceId", "")

    policy_name = extract_resource_name_from_id(resource_id, "firewallPolicies", default="")
    if policy_name:
        context["firewall_policy_name"] = policy_name

    resource_group = extract_resource_name_from_id(resource_id, "resourceGroups", default="")
    if resource_group:
        context["resource_group"] = resource_group

    return context

AnalysisType: rule
Filename: azure_firewall_policy_deleted.py
RuleID: "Azure.MonitorActivity.Network.FirewallPolicyDeleted"
DisplayName: "Azure Firewall Policy Deleted"
Enabled: true
LogTypes:
  - Azure.MonitorActivity
Severity: High
Description: >
  Detects when an Azure Firewall policy is deleted. Firewall policies define critical network security
  rules that control traffic flow and protect resources. Adversaries may delete firewall policies to
  disable network security controls, allow malicious traffic, or enable data exfiltration. This activity
  is a strong indicator of defense evasion or preparation for follow-on attacks.
Reports:
  MITRE ATT&CK:
    - TA0005:T1562.004 # Defense Evasion: Impair Defenses - Disable or Modify System Firewall
Tags:
  - Defense Evasion
  - Impair Defenses
  - Disable or Modify System Firewall
Runbook: |
  1. Query Azure Monitor Activity logs for all network security operations (firewall policy changes, NSG deletions, network watcher deletions) by the callerIpAddress in the 24 hours before and after the alert
  2. Find all firewall policy deletions and modifications in the past 6 hours to determine if this is part of a coordinated attack on network security controls
  3. Check if the callerIpAddress has performed similar network security changes in the past 90 days to establish if this is typical operational activity
Reference: https://github.com/elastic/detection-rules/blob/main/rules/integrations/azure/defense_evasion_azure_firewall_policy_deletion.toml
SummaryAttributes:
  - resourceId
  - callerIpAddress
  - correlationId
  - location

from panther_azuresignin_helpers import actor_user, azure_signin_alert_context, is_sign_in_event


def rule(event):
    if not is_sign_in_event(event):
        return False

    risk_state = event.deep_get("properties", "riskState", default="").lower()
    if risk_state in ["dismissed", "remediated"]:
        return False

    risk_level_during_signin = event.deep_get(
        "properties", "riskLevelDuringSignIn", default=""
    ).lower()
    risk_level_aggregated = event.deep_get("properties", "riskLevelAggregated", default="").lower()

    return risk_level_during_signin == "high" or risk_level_aggregated == "high"


def title(event):
    principal = actor_user(event)
    if principal is None:
        principal = "<NO_PRINCIPALNAME>"

    ip_address = event.deep_get("properties", "ipAddress", default="<UNKNOWN_IP>")

    return f"High-Risk Sign-In Detected: [{principal}] from [{ip_address}]"


def alert_context(event):
    context = azure_signin_alert_context(event)
    return context