Panther rules: aws

from panther_aws_helpers import aws_cloudtrail_success, aws_rule_context

# API calls that are indicative of CloudTrail changes
CLOUDTRAIL_CREATE_UPDATE = {
    "CreateTrail",
    "UpdateTrail",
    "StartLogging",
}


def rule(event):
    return aws_cloudtrail_success(event) and event.get("eventName") in CLOUDTRAIL_CREATE_UPDATE


def title(event):
    return f"CloudTrail [{event.deep_get('requestParameters', 'name')}] was created/updated"


def alert_context(event):
    return aws_rule_context(event)

AnalysisType: rule
Filename: aws_cloudtrail_created.py
RuleID: "AWS.CloudTrail.Created"
DisplayName: "A CloudTrail Was Created or Updated"
Enabled: true
LogTypes:
  - AWS.CloudTrail
Tags:
  - AWS
  - Security Control
  - Discovery:Cloud Service Dashboard
Reports:
  CIS:
    - 3.5
  MITRE ATT&CK:
    - TA0007:T1538

  Stratus Red Team:
    - aws.defense-evasion.cloudtrail-stop
Severity: Info
Description: >
  A CloudTrail Trail was created, updated, or enabled.
Runbook: https://docs.runpanther.io/alert-runbooks/built-in-rules/aws-cloudtrail-modified
Reference: https://docs.aws.amazon.com/awscloudtrail/latest/userguide/cloudtrail-create-and-update-a-trail.html
SummaryAttributes:
  - eventName
  - userAgent
  - sourceIpAddress
  - recipientAccountId
  - p_any_aws_arns

import json
from fnmatch import fnmatch
from unittest.mock import MagicMock

from panther_aws_helpers import aws_cloudtrail_success, aws_rule_context

SECURITY_CONFIG_ACTIONS = {
    "DeleteAccountPublicAccessBlock",
    "DeleteDeliveryChannel",
    "DeleteDetector",
    "DeleteFlowLogs",
    "DeleteRule",
    "DeleteTrail",
    "DisableEbsEncryptionByDefault",
    "DisableRule",
    "StopConfigurationRecorder",
    "StopLogging",
}

ALLOW_LIST = [
    # Add expected events and users here to suppress alerts
    # {"userName": "ExampleUser", "eventName": "DeleteRule"},
]


def rule(event):
    global ALLOW_LIST  # pylint: disable=global-statement
    if isinstance(ALLOW_LIST, MagicMock):
        ALLOW_LIST = json.loads(ALLOW_LIST())  # pylint: disable=not-callable

    if not aws_cloudtrail_success(event):
        return False

    for entry in ALLOW_LIST:
        if fnmatch(
            event.deep_get(
                "userIdentity",
                "sessionContext",
                "sessionIssuer",
                "userName",
                default="",
            ),
            entry["userName"],
        ):
            if fnmatch(event.get("eventName"), entry["eventName"]):
                return False

    if event.get("eventName") == "UpdateDetector":
        return not event.deep_get("requestParameters", "enable", default=True)

    return event.get("eventName") in SECURITY_CONFIG_ACTIONS


def title(event):
    return f"Sensitive AWS API call {event.get('eventName')} made by {event.udm('actor_user')}"


def alert_context(event):
    return aws_rule_context(event)

AnalysisType: rule
Filename: aws_security_configuration_change.py
RuleID: "AWS.CloudTrail.SecurityConfigurationChange"
DisplayName: "Account Security Configuration Changed"
Enabled: true
LogTypes:
  - AWS.CloudTrail
Tags:
  - AWS
  - Defense Evasion:Impair Defenses
Severity: Medium
Reports:
  MITRE ATT&CK:
    - TA0005:T1562

  Stratus Red Team:
    - aws.defense-evasion.cloudtrail-delete
    - aws.defense-evasion.cloudtrail-stop
    - aws.defense-evasion.vpc-remove-flow-logs
Description: An account wide security configuration was changed.
Runbook: >
  Verify that this change was planned. If not, revert the change and update the access control policies to ensure this doesn't happen again.
Reference: https://docs.aws.amazon.com/prescriptive-guidance/latest/aws-startup-security-baseline/controls-acct.html
SummaryAttributes:
  - eventName
  - userAgent
  - sourceIpAddress
  - recipientAccountId
  - p_any_aws_arns

from panther_aws_helpers import aws_cloudtrail_success, aws_rule_context
from panther_detection_helpers.caching import check_account_age


def rule(event):
    # Only check successful ModiyImageAttribute events
    if not aws_cloudtrail_success(event) or event.get("eventName") != "ModifyImageAttribute":
        return False

    added_perms = event.deep_get(
        "requestParameters", "launchPermission", "add", "items", default=[{}]
    )

    for item in added_perms:
        if item.get("group") == "all":
            return True
        if check_account_age(
            item.get("userId", "") + "-" + event.udm("user_account_id", default="")
        ):  # checking if the account is new
            return True

    return False


def alert_context(event):
    return aws_rule_context(event)

AnalysisType: rule
Filename: aws_ami_modified_for_public_access.py
RuleID: "AWS.CloudTrail.AMIModifiedForPublicAccess"
DisplayName: "Amazon Machine Image (AMI) Modified to Allow Public Access"
Enabled: true
LogTypes:
  - AWS.CloudTrail
Tags:
  - AWS
  - Exfiltration:Transfer Data to Cloud Account
Severity: Medium
Reports:
  MITRE ATT&CK:
    - TA0010:T1537
Description: >
  An Amazon Machine Image (AMI) was modified to allow it to be launched by anyone.
  Any sensitive configuration or application data stored in the AMI's block devices is at risk.
Runbook: |
  Determine if the AMI is intended to be publicly accessible.
  If not, first modify the AMI to not be publicly accessible then change any sensitive data stored
  in the block devices associated to the AMI (as they may be compromised).
Reference: https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/sharingamis-intro.html
SummaryAttributes:
  - userAgent
  - sourceIpAddress
  - recipientAccountId
  - p_any_aws_arns

import datetime

from panther_base_helpers import resolve_timestamp_string

TIMEOUT_DAYS = datetime.timedelta(days=90)


def aged_out(timestamp):
    if not timestamp:
        return False
    datetime_ts = resolve_timestamp_string(timestamp)
    if not datetime_ts:
        return False
    return (datetime.datetime.now() - datetime_ts) > TIMEOUT_DAYS


def policy(resource):
    # If a user is less than 4 hours old, it may not have a credential report generated yet.
    # It will be re-scanned periodically until a credential report is found, at which point this
    # policy will be properly evaluated.
    report = resource.get("CredentialReport")
    if not report:
        return True

    if report.get("AccessKey1Active"):
        if aged_out(report.get("AccessKey1LastRotated")):
            return False
    if report.get("AccessKey2Active"):
        if aged_out(report.get("AccessKey2LastRotated")):
            return False

    return True

AnalysisType: policy
Filename: aws_access_key_rotation.py
PolicyID: "AWS.AccessKey.Rotation"
DisplayName: "AWS Access Key Rotation"
Enabled: true
ResourceTypes:
  - AWS.IAM.RootUser
  - AWS.IAM.User
Tags:
  - AWS
  - Identity & Access Management
  - Credential Access:Unsecured Credentials
Reports:
  CIS:
    - 1.4
  MITRE ATT&CK:
    - TA0006:T1552
Severity: Medium
Description: >
  This policy validates that AWS IAM account access keys are rotated every 90 days.
  Rotating access keys will reduce the window of opportunity for an access key that
  is associated with a compromised or terminated account to be used.
Runbook: >
  https://docs.runpanther.io/alert-runbooks/built-in-policies/aws-access-keys-rotated-every-90-days
Reference: >
  https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_access-keys.html

from datetime import timedelta

from panther_base_helpers import deep_get, resolve_timestamp_string

AWS_TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
DEFAULT_TIME = "0001-01-01T00:00:00Z"
MAX_SECONDS_TO_AUTOGEN_KEY = timedelta(seconds=7)


def policy(resource):
    # If a user is less than 4 hours old, it may not have a credential report generated yet.
    # It will be re-scanned periodically until a credential report is found, at which point this
    # policy will be properly evaluated.
    if not resource.get("CredentialReport"):
        return True

    key_rot = deep_get(resource, "CredentialReport", "AccessKey1LastRotated")
    if key_rot == DEFAULT_TIME:
        return True

    create = resource.get("TimeCreated", "")
    key_rot_date = resolve_timestamp_string(key_rot)
    create_date = resolve_timestamp_string(create)

    if not key_rot_date or not create_date:
        return True

    return (key_rot_date - create_date) >= MAX_SECONDS_TO_AUTOGEN_KEY

AnalysisType: policy
Filename: aws_access_keys_at_account_creation.py
DisplayName: "AWS Access Keys At Account Creation"
PolicyID: "AWS.AccessKeys.AccountCreation"
Enabled: true
ResourceTypes:
  - AWS.IAM.User
  - AWS.IAM.RootUser
Tags:
  - AWS
  - Identity & Access Management
Reports:
  CIS:
    - 1.21
Severity: Low
Description: >
  This policy validates that AWS IAM user accounts do not have access keys that were
  created during account creation. This results in excess keys being generated,
  and unnecessary management work in auditing and rotating these keys.
Runbook: >
  https://docs.runpanther.io/alert-runbooks/built-in-policies/aws-access-keys-are-not-created-at-account-creation
Reference: >
  https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_access-keys.html

import datetime

from panther_base_helpers import resolve_timestamp_string

AWS_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
EXPIRATION_BUFFER = datetime.timedelta(days=60)


def policy(resource):
    if not resource.get("NotAfter"):
        return False

    timestamp = resolve_timestamp_string(resource.get("NotAfter"))

    if not timestamp:
        return True

    time_to_expiration = timestamp - datetime.datetime.now()

    return time_to_expiration >= EXPIRATION_BUFFER

AnalysisType: policy
Filename: aws_acm_certificate_expiration.py
PolicyID: "AWS.ACM.Certificate.Expiration"
DisplayName: "AWS ACM Certificate Expiration"
Enabled: true
ResourceTypes:
  - AWS.ACM.Certificate
Reports:
  MITRE ATT&CK:
    - TA0042:T1588
Tags:
  - AWS
  - Availability
  - Resource Development:Obtain Capabilities
Severity: Medium
Description: >
  When a certificate is 60 days away from expiration, ACM automatically attempts to renew it every hour.
Runbook: >
  https://docs.runpanther.io/alert-runbooks/built-in-policies/aws-acm-certificate-is-not-expired
Reference: https://docs.aws.amazon.com/acm/latest/userguide/troubleshooting-renewal.html

def policy(resource):
    if not bool(resource["InUseBy"]):
        return True

    return resource["Status"] == "ISSUED"

AnalysisType: policy
Filename: aws_acm_certificate_valid.py
PolicyID: "AWS.ACM.Certificate.Valid"
DisplayName: "AWS ACM Certificate Status"
Enabled: true
ResourceTypes:
  - AWS.ACM.Certificate
Reports:
  MITRE ATT&CK:
    - TA0042:T1588
Tags:
  - AWS
  - Operations
  - Panther
  - Resource Development:Obtain Capabilities
Severity: High
Description: >
  This policy checks if an ACM certificate renewal is pending or has failed and is in use
  by any other resources within the account.
Runbook: >
  From the ACM panel in the AWS console, check and resolve the certificate status.
Reference: https://docs.aws.amazon.com/acm/latest/userguide/check-certificate-renewal-status.html

# Specify what key and signature algorithms meet your organizations level of security here.
# Included is the AWS ACM default when creating new certificates as an example.
SECURE_KEY_ALGORITHMS = {
    "RSA-2048",  # AWS ACM default
}

SECURE_SIGNATURE_ALGORITHMS = {
    "SHA256WITHRSA",  # AWS ACM default
}


def policy(resource):
    return (
        resource["KeyAlgorithm"] in SECURE_KEY_ALGORITHMS
        and resource["SignatureAlgorithm"] in SECURE_SIGNATURE_ALGORITHMS
    )

AnalysisType: policy
Filename: aws_acm_certificate_has_secure_algorithms.py
PolicyID: "AWS.ACM.HasSecureAlgorithms"
DisplayName: "AWS ACM Secure Algorithms"
Enabled: false
ResourceTypes:
  - AWS.ACM.Certificate
Tags:
  - AWS
  - Configuration Required
  - PCI
  - Defense Evasion:Impair Defenses
Reports:
  PCI:
    - 2.2.3
  MITRE ATT&CK:
    - TA0005:T1562
Severity: High
Description: >
  This policy validates that all ACM certificates are using secure key and signature algorithms.
Runbook: >
  Delete the insecure certificate, and re-create with appropriately secure key and signature algorithms.
Reference: https://docs.aws.amazon.com/acm/latest/userguide/acm-certificate.html#algorithms

AnalysisType: correlation_rule
RuleID: "AWS.Administrative.IAM.User.Created.Group"
DisplayName: "AWS Administrative IAM User Created"
Enabled: false
Severity: Info
Tags:
  - Beta
Description: Identifies when an Administrative IAM user is creates. This could indicate a potential security breach.
Reference: https://stratus-red-team.cloud/attack-techniques/AWS/aws.persistence.iam-create-admin-user/
Reports:
  MITRE ATT&CK:
    - TA0006:T1078
Detection:
    - Group:
        - ID: CreateUser
          RuleID: AWS.IAM.CreateUser
        - ID: AttachAdminUserPolicy
          RuleID: AWS.IAM.AttachAdminUserPolicy
      MatchCriteria:
        field_name:
          - GroupID: CreateUser
            Match: p_alert_context.request_username
          - GroupID: AttachAdminUserPolicy
            Match: p_alert_context.request_username
      LookbackWindowMinutes: 1800
      Schedule:
        RateMinutes: 1440
        TimeoutMinutes: 5

# Add owners as desired
APPROVED_OWNERS = [
    "amazon",
    "microsoft",
]


def policy(resource):
    # These are trusted public snapshot distributors, allow
    if resource["ImageOwnerAlias"] in APPROVED_OWNERS:
        return True

    # Ignore AMIs that are not owned by the scanned account
    if resource["AccountId"] != resource["OwnerId"]:
        return True

    return not resource["Public"]

AnalysisType: policy
Filename: aws_ami_private.py
PolicyID: "AWS.AMI.Private"
DisplayName: "AWS AMI Sharing"
Enabled: true
ResourceTypes:
  - AWS.EC2.AMI
Tags:
  - AWS
  - Panther
  - Panther Enterprise
  - Exfiltration:Transfer Data to Cloud Account
Reports:
  MITRE ATT&CK:
    - TA0010:T1537
Severity: High
Description: >
  This policy ensures that AMIs you have created are not configured to allow public access, which could result in accidental data loss. AMI's that you use but do not own are not evaluated by this policy.
Runbook: >
  Immediately remove public access from the AMI until you can determine whether that setting is intentional
Reference: https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/sharing-amis.html

from panther_base_helpers import deep_get

# MAPPINGS is a dictionary where the Key is an application load balancer ARN, and the
# Value is a WAF web ACL ID. For each Load Balancer ARN present in MAPPINGS,
# this rule verifies that the load balancer has the associated Web ACL
MAPPINGS = {
    "TEST_LOAD_BALANCER_ARN": "TEST_WAF_WEB_ACL_ID",
}


def policy(resource):
    # Check if a Web ACL is required for this load balancer
    if resource["LoadBalancerArn"] not in MAPPINGS:
        return True

    # Check if a Web ACL exists for this load balancer
    if resource["WebAcl"] is None:
        return False

    # Check that the correct Web ACL is assigned for this load balancer
    return deep_get(resource, "WebAcl", "WebACLId") == MAPPINGS[resource["LoadBalancerArn"]]

AnalysisType: policy
Filename: aws_application_load_balancer_web_acl.py
PolicyID: "AWS.ApplicationLoadBalancer.WebACL"
DisplayName: "AWS Application Load Balancer Web ACL"
Enabled: false
ResourceTypes:
  - AWS.ELBV2.ApplicationLoadBalancer
Tags:
  - AWS
  - Configuration Required
  - Security Control
  - Initial Access:Exploit Public-Facing Application
Reports:
  MITRE ATT&CK:
    - TA0001:T1190
Severity: High
Description: >
  This policy validates that all application load balancers have an associated Web ACl to enforce protections against various web attacks.
Runbook: >
  https://docs.runpanther.io/alert-runbooks/built-in-policies/aws-application-load-balancer-has-web-acl
Reference: https://aws.amazon.com/blogs/aws/aws-web-application-firewall-waf-for-application-load-balancers/

from panther_aws_helpers import aws_rule_context


def rule(_):
    return True


def title(event):
    return (
        f"AWS [{event.get('eventName')}] for "
        f"[{event.deep_get('userIdentity', 'arn', default = '<arn_not_found>')}]"
        " from unmanaged IP Address."
    )


def alert_context(event):
    return aws_rule_context(event)

AnalysisType: scheduled_rule
Description: Detects AWS Logins from IP addresses not found in CrowdStrike's AIP list. May indicate unmanaged device being used, or faulty CrowdStrike Sensor.
DisplayName: "AWS Authentication From CrowdStrike Unmanaged Device"
Enabled: false
Filename: aws_authentication_from_crowdstrike_unmanaged_device.py
Reference: https://www.crowdstrike.com/wp-content/uploads/2023/05/crowdstrike-falcon-device-control-data-sheet.pdf
Severity: Medium
DedupPeriodMinutes: 60
RuleID: "AWS.Authentication.From.CrowdStrike.Unmanaged.Device"
Threshold: 1
ScheduledQueries:
  - AWS Authentication from CrowdStrike Unmanaged Device
Tags:
  - Multi-Table Query

AnalysisType: scheduled_query
Description: Detects AWS Authentication events with IP Addresses not found in CrowdStrike's AIP List
Enabled: false
SnowflakeQuery: |
  SELECT *
  FROM panther_logs.public.aws_cloudtrail
  WHERE p_occurs_since('1 hour')
    AND eventName IN ('ConsoleLogin', 'SignIn', 'GetSessionToken')
    AND eventSource IN ('sts.amazonaws.com', 'signin.amazonaws.com')
    AND sourceIPAddress NOT IN
      (
        SELECT DISTINCT aip
        FROM panther_logs.public.crowdstrike_aidmaster
        WHERE p_occurs_since('3 days')
      )

DatabricksQuery: |
  SELECT *
  FROM panther_logs.aws_cloudtrail
  WHERE p_occurs_since('1 hour')
    AND eventName IN ('ConsoleLogin', 'SignIn', 'GetSessionToken')
    AND eventSource IN ('sts.amazonaws.com', 'signin.amazonaws.com')
    AND sourceIPAddress NOT IN
      (
        SELECT DISTINCT aip
        FROM panther_logs.crowdstrike_aidmaster
        WHERE p_occurs_since('3 days')
      )
QueryName: "AWS Authentication from CrowdStrike Unmanaged Device"
Schedule:
  RateMinutes: 60
  TimeoutMinutes: 3
Tags:
  - Multi-Table Query

# This file is the part of the Crowdstrike FDREvent migration, and it's the equivalent of
# https://github.com/panther-labs/panther-analysis/blob/b61db1ecf3967c5f6a44c1782f8891fd5f54384d/queries/aws_queries/AWS_Authentication_from_CrowdStrike_Unmanaged_Device.yml
#
AnalysisType: scheduled_query
Description: Detects AWS Authentication events with IP Addresses not found in CrowdStrike's AIP List
Enabled: false
SnowflakeQuery: |
  SELECT *
  FROM panther_logs.public.aws_cloudtrail
  WHERE p_occurs_since('1 days')
    AND eventName IN ('ConsoleLogin', 'SignIn', 'GetSessionToken')
    AND eventSource IN ('sts.amazonaws.com', 'signin.amazonaws.com')
    AND sourceIPAddress NOT IN
      (
        SELECT DISTINCT aip
        FROM panther_logs.public.crowdstrike_fdrevent
        WHERE p_occurs_since('3 days') AND panther_logs.public.crowdstrike_fdrevent.fdr_event_type = 'aid_master'
      )

DatabricksQuery: |
  SELECT *
  FROM panther_logs.aws_cloudtrail
  WHERE p_occurs_since('1 days')
    AND eventName IN ('ConsoleLogin', 'SignIn', 'GetSessionToken')
    AND eventSource IN ('sts.amazonaws.com', 'signin.amazonaws.com')
    AND sourceIPAddress NOT IN
      (
        SELECT DISTINCT aip
        FROM panther_logs.crowdstrike_fdrevent
        WHERE p_occurs_since('3 days') AND panther_logs.crowdstrike_fdrevent.fdr_event_type = 'aid_master'
      )
QueryName: "AWS Authentication from CrowdStrike Unmanaged Device (crowdstrike_fdrevent table)"
Schedule:
  RateMinutes: 1440
  TimeoutMinutes: 3
Tags:
  - Multi-Table Query

AnalysisType: correlation_rule
RuleID: "AWS.Backdoor.Administrative.IAM.Role.Created.Group"
DisplayName: "AWS Backdoor Administrative IAM Role Created"
Enabled: false
Severity: High
Description: Identifies when CreateRole and AttachAdminRolePolicy CloudTrail events occur in a short period of time. This sequence could indicate a potential security breach.
Reference: https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies.html
Reports:
  MITRE ATT&CK:
    - TA0007:T1078
Detection:
    - Group:
        - ID: CreateRole
          RuleID: AWS.IAM.CreateRole
        - ID: AttachAdminRolePolicy
          RuleID: AWS.IAM.AttachAdminRolePolicy
      MatchCriteria:
        field_name:
          - GroupID: CreateRole
            Match: p_alert_context.request_rolename
          - GroupID: AttachAdminRolePolicy
            Match: p_alert_context.request_rolename
      LookbackWindowMinutes: 1800
      Schedule:
        RateMinutes: 1440
        TimeoutMinutes: 5

from panther_aws_helpers import aws_cloudtrail_success, aws_rule_context

GUARDRAIL_EVENTS = {"DeleteGuardrail", "UpdateGuardrail"}


def rule(event):
    if (
        event.get("eventSource") == "bedrock.amazonaws.com"
        and event.get("eventName") in GUARDRAIL_EVENTS
        and aws_cloudtrail_success(event)
    ):
        return True
    return False


def title(event):
    user = event.udm("actor_user")
    guardrail = event.deep_get("requestParameters", "guardrailIdentifier")
    action = event.get("eventName").replace("Guardrail", "").lower()
    return f"User [{user}] {action}d Bedrock guardrail [{guardrail}]"


def severity(event):
    if event.get("eventName") == "UpdateGuardrail":
        return "LOW"
    return "DEFAULT"


def alert_context(event):
    return aws_rule_context(event)

AnalysisType: rule
Filename: aws_bedrock_guardrail_update_delete.py
RuleID: "AWS.Bedrock.GuardrailUpdateDelete"
DisplayName: "AWS Bedrock Guardrail Updated or Deleted"
Enabled: true
LogTypes:
  - AWS.CloudTrail
Tags:
  - AWS
  - Bedrock
  - Generative AI Guardrails
  - AML.T0054
  - LLM Jailbreak
  - "Impair Defenses: Disable or Modify Tools"
  - Defense Evasion
Severity: Medium
Reports:
  MITRE ATT&CK:
    - TA0005:T1562.001 # Impair Defenses: Disable or Modify Tools
Description: >
  An Amazon Bedrock Guardrail was updated or deleted.
  Amazon Bedrock Guardrails are used to implement application-specific safeguards based on your use cases and responsible AI policies.
  Updating or deleting a guardrail can have security implications to your AI workloads.
Runbook: |
  Review the guardrail update or deletion to ensure that it was authorized and that it does not introduce security risks to your AI workloads.
  If the guardrail update or deletion was unauthorized, investigate the incident and take appropriate action.
  https://atlas.mitre.org/mitigations/AML.M0020
Reference: https://docs.aws.amazon.com/bedrock/latest/APIReference/API_DeleteGuardrail.html
SummaryAttributes:
  - userAgent
  - sourceIpAddress
  - recipientAccountId
  - p_any_aws_arns

def rule(event):
    # Only process InvokeModel and Converse operations
    if event.get("operation") not in ["InvokeModel", "Converse"]:
        return False

    # retrieve the necessary values from the logs
    token_usage = event.deep_get("output", "outputBodyJson", "usage", "totalTokens", default=0)
    model_id = event.get("modelId", default="")
    # Get the appropriate threshold for each model
    if "haiku" in model_id:
        threshold = 3000
    elif "sonnet" in model_id:
        threshold = 4000
    elif "opus" in model_id:
        threshold = 5000
    else:
        threshold = 4000  # default threshold
    # Check for abnormal token usage
    if token_usage > threshold:
        return True
    # Flag unusual token patterns (high usage with no actual output)
    output_tokens = event.deep_get("output", "outputBodyJson", "usage", "outputTokens", default=0)
    if token_usage > 1000 and output_tokens == 0:
        return True
    return False


def title(event):
    model_id = event.get("modelId", default="unknown")
    operation_name = event.get("operation", default="unknown")
    account_id = event.get("accountId", default="unknown")
    token_usage = event.deep_get("output", "outputBodyJson", "usage", "totalTokens", default=0)
    title_parts = [
        f"Abnormal token usage detected: {token_usage} tokens",
        f"Model: {model_id}",
        f"Operation: {operation_name}",
        f"Account: {account_id}",
    ]
    return " | ".join(title_parts)

AnalysisType: rule
Filename: aws_bedrockmodelinvocation_abnormaltokenusage.py
RuleID: "AWS.BedrockModelInvocation.AbnormalTokenUsage"
DisplayName: "AWS Bedrock Model Invocation Abnormal Token Usage"
Enabled: true
LogTypes:
    - AWS.BedrockModelInvocation
Tags:
    - AWS
    - Bedrock
    - Resource Hijacking
Status: Experimental
Severity: Info
Reports:
    MITRE ATT&CK:
        - TA0040:T1496.004
Description: Monitors for potential misuse or abuse of AWS Bedrock AI models by detecting abnormal token usage patterns and alerts when the total token usage exceeds the appropriate threshold for each different type of model.
Runbook: Verify the alert details by checking token usage, model ID, and account information to confirm unusual activity, examine user access patterns to identify potential credential compromise, and look for evidence of prompt injection, unusual repetition, or attempts to bypass usage limits. Apply stricter usage quotas to the affected account, block suspicious IP addresses, and enhance the guardrails that are in place.
DedupPeriodMinutes: 60
Threshold: 1
Reference: https://stratus-red-team.cloud/attack-techniques/AWS/aws.impact.bedrock-invoke-model/
SummaryAttributes:
  - p_any_aws_account_ids
  - p_any_aws_arns
InlineFilters:
    - All: []

def rule(event):
    if event.get("operation") != "InvokeModel" and event.get("operation") != "Converse":
        return False

    stop_reason = event.deep_get(
        "output", "outputBodyJSON", "stopReason", default="<UNKNOWN REASON>"
    )
    action_reason = event.deep_get(
        "output",
        "outputBodyJSON",
        "amazon-bedrock-trace",
        "guardrail",
        "actionReason",
        default="<UNKNOWN ACTION REASON>",
    )
    return stop_reason == "guardrail_intervened" or action_reason.startswith("Guardrail blocked")


def title(event):
    model_id = event.get("modelId")
    operation_name = event.get("operation")
    account_id = event.get("accountId")
    stop_reason = event.deep_get(
        "output", "outputBodyJSON", "stopReason", default="<UNKNOWN REASON>"
    )
    action_reason = event.deep_get(
        "output",
        "outputBodyJSON",
        "amazon-bedrock-trace",
        "guardrail",
        "actionReason",
        default="<UNKNOWN ACTION REASON>",
    )
    if action_reason == "<UNKNOWN ACTION REASON>":
        return (
            f"The model [{model_id}] was invoked with the operation [{operation_name}] "
            f"by the account [{account_id}]. Stop reason [{stop_reason}]."
        )
    if stop_reason == "<UNKNOWN REASON>":
        return (
            f"The model [{model_id}] was invoked with the operation [{operation_name}] "
            f"by the account [{account_id}]. Action reason [{action_reason}]."
        )
    # Handle the case when both values are known
    return (
        f"The model [{model_id}] was invoked with the operation [{operation_name}] "
        f"by the account [{account_id}]. Stop reason [{stop_reason}]. "
        f"Action reason [{action_reason}]."
    )

AnalysisType: rule
Filename: aws_bedrockmodelinvocation_guardrailintervened.py
RuleID: "AWS.BedrockModelInvocation.GuardRailIntervened"
DisplayName: "AWS Bedrock Model Invocation GuardRail Intervened"
Enabled: true
LogTypes:
    - AWS.BedrockModelInvocation
Tags:
    - AWS
    - Bedrock
    - Persistence
    - Manipulate AI Model
Status: Experimental
Severity: Info
Reports:
    MITRE ATT&CK:
        - TA0006:T0018.000
Description: Detects when AWS Bedrock guardrail features have intervened during AI model invocations. It specifically monitors when an AI model request was blocked by Guardrails. This helps security teams identify when users attempt to generate potentially harmful or inappropriate content through AWS Bedrock models.
Runbook: Confirm alert details by reviewing the model ID, operation name, account ID, and the specific guardrail intervention reasons provided in the alert description. Analyze the user prompts that triggered the guardrail by examining the Bedrock console logs for the associated requestId, looking for patterns of attempted model poisoning or prompt injection techniques. If suspicious activity is confirmed, temporarily restrict the access of the malicious actor to Bedrock services, preserve all evidence of the interaction, and escalate to the security team for further analysis of potential AI model manipulation attempts. https://atlas.mitre.org/mitigations/AML.M0005
DedupPeriodMinutes: 60
Threshold: 1
Reference: https://stratus-red-team.cloud/attack-techniques/AWS/aws.impact.bedrock-invoke-model/, https://docs.aws.amazon.com/bedrock/latest/userguide/guardrails.html
SummaryAttributes:
  - p_any_aws_account_ids
  - p_any_aws_arns
InlineFilters:
    - All: []

from panther_aws_helpers import aws_cloudtrail_success, aws_rule_context


def rule(event):
    if (
        event.get("eventSource") == "bedrock.amazonaws.com"
        and event.get("eventName") == "DeleteModelInvocationLoggingConfiguration"
        and aws_cloudtrail_success(event)
    ):
        return True
    return False


def title(event):
    user = event.udm("actor_user")
    return f"User [{user}] deleted Bedrock model invocation logging configuration"


def alert_context(event):
    return aws_rule_context(event)

AnalysisType: rule
Filename: aws_bedrock_deletemodelinvocationloggingconfiguration.py
RuleID: "AWS.Bedrock.DeleteModelInvocationLoggingConfiguration"
DisplayName: "AWS Bedrock Model Invocation Logging Configuration Deleted"
Enabled: true
LogTypes:
  - AWS.CloudTrail
Tags:
  - AWS
  - Bedrock
  - "Impair Defenses: Impair Command History Logging"
  - Defense Evastion
Severity: Medium
Reports:
  MITRE ATT&CK:
    - TA0005:T1562.003 # Impair Defenses: Impair Command History Logging
Description: >
  An Amazon Bedrock Model Invocation Logging Configuration was deleted.
  Use model invocation logging to collect metadata, requests, and responses for all model invocations in your account.
  Deleting a model invocation logging configuration can have security implications to your AI workloads.
Runbook: |
  Review the model invocation logging configuration deletion to ensure that it was authorized and that it does not introduce security risks to your AI workloads.
  If the model invocation logging configuration deletion was unauthorized, investigate the incident and take appropriate action.
Reference: https://docs.aws.amazon.com/bedrock/latest/userguide/model-invocation-logging.html
SummaryAttributes:
  - userAgent
  - sourceIpAddress
  - recipientAccountId
  - p_any_aws_arns

def policy(resource):

    return bool(resource["Encrypted"])

AnalysisType: policy
Filename: aws_ec2_cde_volume_encrypted.py
PolicyID: "AWS.EC2.CDEVolumeEncrypted"
DisplayName: "AWS CDE EC2 Volume Encryption"
Enabled: false
ResourceTypes:
  - AWS.EC2.Volume
Tags:
  - AWS
  - PCI
  - Collection:Data From Local System
Reports:
  PCI:
    - 3.4
  MITRE ATT&CK:
    - TA0009:T1005
Severity: Medium
Description: >
  This policy ensures that all EC2 volumes that contain CDE are encrypted.
  Be sure to configure CDE definitions before enabling this policy.
Runbook: >
  https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSEncryption.html#encryption-by-default-api
Reference: https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSEncryption.html

from panther_base_helpers import deep_get

# CloudFormation stacks tagged with "STACK=(name)" will be marked as passing.
IGNORE_STACK_TAGS = {
    "panther-bootstrap-gateway",
    "panther-cloud-security",
    "panther-core",
    "panther-log-analysis",
}


def policy(resource):
    if deep_get(resource, "DriftInformation", "StackDriftStatus") != "DRIFTED":
        return True

    # Some of Panther's own stacks contain Lambda functions which will always show as "drifted."
    # Panther stacks have a fixed "Stack" tag, even though the real stack name is dynamic.
    tags = resource["Tags"]
    if tags.get("Application") == "Panther" and tags.get("Stack") in IGNORE_STACK_TAGS:
        return True

    return False

AnalysisType: policy
Filename: aws_cloudformation_stack_drifted.py
PolicyID: "AWS.CloudFormation.Stack.Drifted"
DisplayName: "AWS CloudFormation Stack Drift"
Enabled: true
ResourceTypes:
  - AWS.CloudFormation.Stack
Reports:
  MITRE ATT&CK:
    - TA0040:T1496
Tags:
  - AWS
  - Operations
  - Panther
  - Impact:Resource Hijacking
Severity: Low
Description: >
  A stack has drifted from its defined configuration.
Runbook: |
  From the CloudFormation web console, look at the drifted resources for the failing stack.
  If the drift is expected, update the policy ignore list to exclude this stack.
  Otherwise, analyze CloudTrail logs to understand who changed the drifted resource(s) and ensure
  it was legitimate access.
Reference: https://amzn.to/2z8dDFW

def policy(resource):
    # Ignore stack sets as this setting cannot be set on those
    if resource["Name"].startswith("StackSet-"):
        return True

    return resource["RoleARN"] is not None

AnalysisType: policy
Filename: aws_cloudformation_stack_uses_iam_role.py
PolicyID: "AWS.CloudFormation.Stack.UsesIAMServiceRole"
DisplayName: "AWS CloudFormation Stack IAM Service Role"
Enabled: true
ResourceTypes:
  - AWS.CloudFormation.Stack
Tags:
  - AWS
  - Operations
  - Panther
Severity: Info
Description: >
  Associating IAM roles with CloudFormation stacks ensures least privilege when making
  changes to your account.
Runbook: >
  Create a new IAM role to be assumable by the cloudformation service, and then update the
  current stack with the --role-arn argument.
Reference: https://amzn.to/2HAdfny

def policy(resource):
    # On nested stacks, this can only be set on the root stack
    if resource["RootId"] is not None:
        return True

    return resource["EnableTerminationProtection"] is True

AnalysisType: policy
Filename: aws_cloudformation_termination_protection.py
PolicyID: "AWS.CloudFormation.Stack.TerminationProtection"
DisplayName: "AWS CloudFormation Stack Termination Protection"
Enabled: true
ResourceTypes:
  - AWS.CloudFormation.Stack
Reports:
  MITRE ATT&CK:
    - TA0040:T1496
Tags:
  - AWS
  - Operations
  - Panther
  - Impact:Resource Hijacking
Severity: Info
Description: >
  Protects a CloudFormation stack from accidentally being deleted. If you attempt to delete a stack
  with termination protection enabled, the deletion fails and the stack, including its status,
  will remain unchanged.
Runbook: |
  This setting can only be enabled on stack creation. To add it to an existing stack, it must
  be re-created with the --enable-termination-protection flag.
Reference: https://amzn.to/2HAdfny