CredentialCompromiseDetection-Kinesis-Template.yaml

AWSTemplateFormatVersion: '2010-09-09'
Description: Template to detect credential compromise via CloudWatch Events and expected IPs. What's in Your CloudWatch?

Parameters:

  pDebug:
    Description: Enable Debugging in CloudWatchLogs
    Type: String
    Default: False

  pObjectKey:
    Description: Name of the S3 Object in the bucket with the expected IPs.
    Type: String
    Default: Instance-Data.json

  pReservedConcurrentExecutions:
    Description: Number of concurrent executions for the detect functions
    Type: Number
    Default: 100  # 10% of default limit of 1000

  pTopicArn:
    Description: ARN of the Topic which to send events to
    Type: String

  pTopicRegion:
    Description: Region where the Topic is.
    Type: String

  pCloudTrailLogGroupName:
    Description: LogGroup in CloudWatch logs to get the CloudTrail events
    Type: String
    Default: "CloudTrail/DefaultLogGroup"

Resources:

  Bucket:
    Type: AWS::S3::Bucket
    Properties:
      AccessControl: Private
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: AES256
      PublicAccessBlockConfiguration:
        BlockPublicAcls: True
        BlockPublicPolicy: True
        IgnorePublicAcls: True
        RestrictPublicBuckets: True

  LambdaRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Principal:
            Service:
            - lambda.amazonaws.com
          Action:
          - sts:AssumeRole
      Path: /
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonEC2ReadOnlyAccess
      Policies:
      - PolicyName: S3Access
        PolicyDocument:
          Version: '2012-10-17'
          Statement:
          - Action:
            - s3:*
            Effect: Allow
            Resource:
              - !Join ['', ['arn:aws:s3:::', !Ref Bucket , /*]]
              - !Join ['', ['arn:aws:s3:::', !Ref Bucket ]]
          - Action:
            - s3:ListAllMyBuckets
            - s3:GetBucketLocation
            Effect: Allow
            Resource: '*'
      - PolicyName: LambdaLogging
        PolicyDocument:
          Version: '2012-10-17'
          Statement:
          - Resource: '*'
            Action:
            - logs:*
            Effect: Allow
      # - PolicyName: GetMessages
      #   PolicyDocument:
      #     Version: '2012-10-17'
      #     Statement:
      #     - Resource: !GetAtt CloudWatchEventQueue.Arn
      #       Action:
      #       - sqs:*
      #       Effect: Allow
      - PolicyName: PublishEvents
        PolicyDocument:
          Version: '2012-10-17'
          Statement:
          - Resource: !Ref pTopicArn
            Action:
            - sns:publish
            Effect: Allow
      - PolicyName: ReadKineses
        PolicyDocument:
          Version: '2012-10-17'
          Statement:
          - Resource: !GetAtt CloudTrailStream.Arn
            Effect: Allow
            Action:
            - kinesis:DescribeStream
            - kinesis:DescribeStreamSummary
            - kinesis:GetRecords
            - kinesis:GetShardIterator
            - kinesis:ListShards
            - kinesis:ListStreams
            - kinesis:SubscribeToShard
            - logs:CreateLogGroup
            - logs:CreateLogStream
            - logs:PutLogEvent

  DetectFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: !Sub "${AWS::StackName}-detect"
      Description: Compare Cloudtrail Event Source IP to expected values
      Handler: index.handler
      Runtime: python3.6
      Timeout: 60
      ReservedConcurrentExecutions: !Ref pReservedConcurrentExecutions
      Role: !GetAtt LambdaRole.Arn
      Code:
        ZipFile: !Sub |
          import boto3
          import re
          import json
          import os
          import base64
          import gzip

          import logging
          logger = logging.getLogger()
          logger.setLevel(logging.INFO)
          logging.getLogger('botocore').setLevel(logging.WARNING)
          logging.getLogger('boto3').setLevel(logging.WARNING)


          def handler(event, context):
            re_principal = re.compile(r'AR[^\:]+\:i\-[0-9a-z]+')
            instance_data = []
            logger.debug(f"Event: {event}")

            try:
              logger.info(f"Received {len(event['Records'])} kinesis events")
              for record in event['Records']:
                cwl_str = gzip.decompress(base64.b64decode(record["kinesis"]["data"]))
                logger.debug(f"Log String: {cwl_str}")
                cwl_records = json.loads(cwl_str)
                if cwl_records['messageType'] != "DATA_MESSAGE":
                  continue

                logger.info(f"Received {len(cwl_records['logEvents'])} CW Log events")
                for log_event in cwl_records['logEvents']:
                  cwevent = json.loads(log_event['message'])

                if cwevent['eventType'] != "AwsApiCall":
                  logger.debug(f"Got Event type that should have been filtered: {cwevent}")
                  continue

                # Filter out stuff
                if "amazonaws.com" in cwevent['sourceIPAddress']:
                  continue

                logger.debug("cwevent: {}".format(json.dumps(cwevent, sort_keys=True)))

                principal = cwevent['userIdentity']['principalId']
                if not re_principal.match(principal):
                  logger.debug(f"Principal {principal} is not an instance profile role. Skipping")
                  continue

                if instance_data is not None:  # Wait to get instance data till we need it
                  instance_data = get_instance_data(os.environ['BUCKET'], os.environ['OBJECT'])

                  # Extract the instance Id
                  (role_id, instance_id) = principal.split(":")
                  source_ip = cwevent['sourceIPAddress']
                  region = cwevent['awsRegion']

                  if instance_id not in instance_data['instances']:
                    logger.info(f"Instance {instance_id} is not in my database of instances")
                    send_event("InstanceMissing", cwevent, f"Instance {instance_id} is not in my database of instances", instance_id, source_ip, [])
                    continue

                  if source_ip in instance_data['instances'][instance_id]:
                    logger.debug(f"Event is from expected IP: {cwevent}")
                    continue

                  if source_ip in instance_data['all_ips']:
                    logger.info(f"Event is not from expected ip, but from an account IP: {cwevent}")
                    continue

                  logger.error(f"Event from IP {source_ip} for instance {instance_id} is not from expected addresses: {cwevent}")
                  send_event("BadSource", cwevent, f"Event from IP {source_ip} for instance {instance_id} is not from expected addresses", instance_id, source_ip, instance_data['instances'][instance_id])
                  return(200)
            except Exception as e:
              logger.critical(f"Exception {e} processing")
              raise

          def send_event(eventtype, cwevent, message, instance_id, source_ip, expected_ips):
            try:
              client = boto3.client('sns', region_name=os.environ['TOPIC_REGION'])
              snsmessage = {
              'type': eventtype,
              'CloudTrailEvent': cwevent,
              'message': message,
              'expected_ips': expected_ips,
              'instance_id': instance_id,
              'uniq_id': f"{instance_id}-{source_ip}"     # We can dedup events on this later
              }
              response = client.publish(
                TopicArn=os.environ['TOPICARN'],
                Message=json.dumps(snsmessage),
                Subject=f'Potential Credential Compromise for {instance_id}',
                )
            except Exception as e:
              logger.critical(f"Unable to send message to SNS: {e} \n{snsmessage}")
              raise

          def get_instance_data(bucket, obj_key):
            '''get the object to index from S3 and return the parsed json'''
            s3 = boto3.client('s3')
            response = s3.get_object(
              Bucket=bucket,
              Key=obj_key
              )
            return(json.loads(response['Body'].read()))
          ### END OF CODE ###

      Environment:
        Variables:
          BUCKET: !Ref Bucket
          OBJECT: !Ref pObjectKey
          DEBUG: !Ref pDebug
          TOPICARN: !Ref pTopicArn
          TOPIC_REGION: !Ref pTopicRegion
      # Tags inherited from Stack

  InventoryFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: !Sub "${AWS::StackName}-inventory"
      Description: Collect Instance IPs and NatGateway Addresses
      Handler: index.handler
      Runtime: python3.6
      Timeout: 150
      MemorySize: 768
      Role: !GetAtt LambdaRole.Arn
      Code:
        ZipFile: !Sub |
          import boto3
          from botocore.exceptions import ClientError
          import json
          import os
          import datetime

          import logging
          logger = logging.getLogger()
          logger.setLevel(logging.INFO)
          logging.getLogger('botocore').setLevel(logging.WARNING)
          logging.getLogger('boto3').setLevel(logging.WARNING)


          def handler(event, context):

              output_json = {}
              output_json['instances'] = {}
              output_json['all_ips'] = []
              output_json['nat_gateways'] = {}

              try:
                  ec2_client = boto3.client('ec2')

                  interfaces = get_all_interfaces(ec2_client)
                  for eni in interfaces:
                      if eni['InterfaceType'] != 'nat_gateway':
                          continue
                      if 'PublicIp' not in eni['Association']:
                          continue

                      if eni['VpcId'] not in output_json['nat_gateways']:  # make it an array
                          output_json['nat_gateways'][eni['VpcId']] = []

                      output_json['nat_gateways'][eni['VpcId']].append(eni['Association']['PublicIp'])
                      output_json['all_ips'].append(eni['Association']['PublicIp'])

                  for reservation in get_all_instances(ec2_client):
                      for instance in reservation['Instances']:
                          instance_id = instance['InstanceId']
                          vpc_id = instance['VpcId']
                          output_json['instances'][instance_id] = []

                          if 'PublicIpAddress' in instance:
                              public_ip = instance['PublicIpAddress']
                              output_json['instances'][instance_id].append(public_ip)
                              output_json['all_ips'].append(public_ip)

                          if vpc_id in output_json['nat_gateways']:
                              output_json['instances'][instance_id] += output_json['nat_gateways'][vpc_id]

                  output_json['data_collected'] = str(datetime.datetime.now())

              except ClientError as e:
                  logger.critical("AWS Error getting info: {}".format(e))
                  raise
              except Exception as e:
                  logger.critical("{}".format(e))
                  raise

              s3client = boto3.client('s3')
              try:
                  s3client.put_object(
                      Body=json.dumps(output_json, sort_keys=True, default=str, indent=2),
                      Bucket=os.environ['BUCKET'],
                      ContentType='application/json',
                      Key=os.environ['OBJECT'],
                  )
                  return(event)
              except ClientError as e:
                  logger.error("Unable to save object {}: {}".format(os.environ['OBJECT'], e))
                  raise

          def get_all_instances(ec2_client):
              output = []
              filters = [{'Name': 'instance-state-name', 'Values': ['running']}]
              response = ec2_client.describe_instances(Filters=filters)
              while 'NextToken' in response:
                  output += response['Reservations']
                  response = ec2_client.describe_instances(Filters=filters, NextToken=response['NextToken'])
              output += response['Reservations']
              return(output)


          def get_all_interfaces(ec2_client):
              interfaces = []
              response = ec2_client.describe_network_interfaces()
              while 'NextToken' in response:  # Gotta Catch 'em all!
                  interfaces += response['NetworkInterfaces']
                  response = ec2_client.describe_network_interfaces(NextToken=response['NextToken'])
              interfaces += response['NetworkInterfaces']
              return(interfaces)
          ### END OF CODE ###
      Environment:
        Variables:
          BUCKET: !Ref Bucket
          OBJECT: !Ref pObjectKey
          DEBUG: !Ref pDebug
      # Tags inherited from Stack

  CloudTrailStream:
    Type: AWS::Kinesis::Stream
    Properties:
      RetentionPeriodHours: 24
      ShardCount: 1
      StreamEncryption:
        EncryptionType: KMS
        KeyId: alias/aws/kinesis

  CloudTrailSubscriptionFilter:
    Type: AWS::Logs::SubscriptionFilter
    Properties:
      DestinationArn: !GetAtt CloudTrailStream.Arn
      FilterPattern: "{($.userIdentity.type = AssumedRole) && ($.eventType = AwsApiCall) && ($.sourceIPAddress != *.amazonaws.com)}"
      LogGroupName: !Ref pCloudTrailLogGroupName
      RoleArn: !GetAtt LogsInvokeKinesisRole.Arn

  LogsInvokeKinesisRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Principal:
            Service:
            - events.amazonaws.com
            - !Sub logs.${AWS::Region}.amazonaws.com
          Action:
          - sts:AssumeRole
      Path: /
      Policies:
      - PolicyName: KinesisPut
        PolicyDocument:
          Version: '2012-10-17'
          Statement:
          - Action:
            - kinesis:PutRecord
            - kinesis:PutRecords
            Effect: Allow
            Resource:
              - !GetAtt CloudTrailStream.Arn

  DetectFunctionMapping:
    Type: AWS::Lambda::EventSourceMapping
    Properties:
      BatchSize: 20
      Enabled: True # FIXME !Ref pState
      EventSourceArn: !GetAtt CloudTrailStream.Arn
      FunctionName: !GetAtt DetectFunction.Arn
      StartingPosition: LATEST

  TriggerInventoryRule:
    Type: "AWS::Events::Rule"
    Properties:
      Description: !Sub "${AWS::StackName} Trigger gathering of IP Addresses"
      ScheduleExpression: rate(10 minutes)
      Targets:
      - Arn: !GetAtt InventoryFunction.Arn
        Id: TargetFunctionV1

  InventoryLambdaPermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !GetAtt InventoryFunction.Arn
      Principal: events.amazonaws.com
      SourceArn: !GetAtt TriggerInventoryRule.Arn
      Action: lambda:invokeFunction

Outputs:
  StackName:
    Description: Name of this Stack
    Value: !Ref AWS::StackName

  TemplateVersion:
    Description: Version of this CFT
    Value: 0.0.1