Signal - VPC Flow Logs Allowed SSH

AnalysisType: rule
Filename: aws_vpc_ssh_allowed_signal.py
RuleID: "AWS.VPC.SSHAllowedSignal"
DisplayName: "Signal - VPC Flow Logs Allowed SSH"
Enabled: true
CreateAlert: false
LogTypes:
  - AWS.VPCFlow
Tags:
  - AWS
  - Signal
Reports:
  MITRE ATT&CK:
    - TA0008:T1021.004 # Lateral Movement: Remote Services: SSH
Severity: Info
Description: >
  VPC Flow Logs observed inbound traffic on SSH port.
  This rule is a signal to be used in correlation rules.
Tests:
  - Name: Public to Private IP on SSH Allowed
    ExpectedResult: true
    Log:
      {
        "dstPort": 22,
        "dstAddr": "10.0.0.1",
        "srcAddr": "1.1.1.1",
        "instanceId": "i-0d4c7318592c6a2c7",
        "action": "ACCEPT",
        "p_log_type": "AWS.VPCFlow",
      }
  - Name: Public to Private IP on non-SSH
    ExpectedResult: false
    Log:
      {
        "dstPort": 443,
        "dstAddr": "10.0.0.1",
        "srcAddr": "1.1.1.1",
        "instanceId": "i-0d4c7318592c6a2c7",
        "action": "ACCEPT",
        "p_log_type": "AWS.VPCFlow",
      }
  - Name: Private to Private IP on SSH
    ExpectedResult: false
    Log:
      {
        "dstPort": 22,
        "dstAddr": "10.0.0.1",
        "srcAddr": "10.10.10.1",
        "instanceId": "i-0d4c7318592c6a2c7",
        "action": "ACCEPT",
        "p_log_type": "AWS.VPCFlow",
      }
  - Name: Public to Private IP on SSH Denied
    ExpectedResult: false
    Log:
      {
        "dstPort": 22,
        "dstAddr": "10.0.0.1",
        "srcAddr": "1.1.1.1",
        "instanceId": "i-0d4c7318592c6a2c7",
        "action": "REJECT",
        "p_log_type": "AWS.VPCFlow",
      }


# ------ paired body: aws_vpc_ssh_allowed_signal.py ------

from ipaddress import ip_network

from panther_aws_helpers import aws_rule_context


def rule(event):
    # Defaults to True (no alert) if 'dstport' is not present
    if event.udm("destination_port") != 22 or event.get("action") != "ACCEPT":
        return False

    # Only monitor for traffic coming from non-private IP space
    #
    # Defaults to True (no alert) if 'srcaddr' key is not present
    source_ip = event.udm("source_ip") or "0.0.0.0/32"
    if not ip_network(source_ip).is_global:
        return False

    # Alert if the traffic is destined for internal IP addresses
    #
    # Defaults to False(no alert) if 'dstaddr' key is not present
    destination_ip = event.udm("destination_ip") or "1.0.0.0/32"
    return not ip_network(destination_ip).is_global


def alert_context(event):
    return aws_rule_context(event)