As part of the conversation in the last article. Tracking could be expansive, consisting of multiple resources of info. As a DevOps or Cloud Designer, we may need to keep an eye on some sources which could not easily obtain from AWS Config.
For instance:
- Exactly how to monitor customized application logs?
- How to keep an eye on AWS solution metrics like RDB link numbers within a period?
- Just how to be extra flexble in checking AWS solution API occasions?
Introduction
To resolve above concerns, we may require to incorporate AWS Cloudtrail + AWS Cloudwatch AWS will tape-record its occasions in AWS Cloudtrail, and AWS Cloudwatch have log group that might keep those logs, or your application logs.
What AWS solutions could be made use of:
Amazon CloudWatch — collection and aggregation of metrics/logs
AWS Lambda — run code in reaction to occasions
Amazon SNS topic– a taken care of messaging solution for notice
Amazon CloudTrail — collection of AWS API event
In below layout, AWS Cloudtrail will stream AWS API events to the Cloudwatch and shop in log team. We arrangement the filter pattern to filter the keywords in the logs, and subscribe the personalized events to the examine lambda, which the logic will certainly define if that event is compliant or otherwise.
Removal will certainly occur as soon as the event is defined as non-compliant. All non-compliant events will certainly sent to SNS as notice along with to all subscirbers.
Diagram
Let’s Start
We are making use of terraform to deploy all called for services, which includes: AWS Config, AWS Config Recorder, AWS Config Policy, S 3 and Lambda, and all IAM resources it calls for.
This instance will certainly filter the occasions from Cloudwatch log teams, and Lambda will certainly push notification once any VPC or EC 2 have public accessibility.
Above is the main.tf and it will develop AWS SNS subject for notification and CloudWatch log group.
Currently, listed below are lambda and Cloudtrail template.
You will certainly need to configure the variables to deploy those design templates. After release, it will develop all the sources and required role for CloudWatch documents.
lambda for occasions:
import json
import os
import datetime
import zlib
import json
from base 64 import b 64 decode
from client.aws import get_client
from rules.vpc import evaluate_vpc
APPLICABLE_RESOURCES = [
"AWS::EC2::NetworkInterface",
"AWS::EC2::SecurityGroup",
"AWS::EC2::NatGateway",
"AWS::EC2::EgressOnlyInternetGateway",
"AWS::EC2::VPCEndpoint",
"AWS::EC2::VPCEndpointService",
"AWS::EC2::VPCPeeringConnection",
"AWS::EC2::InternetGateway",
"AWS::EC2::RouteTable",
"AWS::EC2::VPC",
"AWS::EC2::Subnet",
"AWS::EC2::VPNGateway",
]
def check_defined(recommendation, reference_name):
if not referral:
raise Exception("Mistake: ", reference_name, "is not specified")
return referral
# Examine whether the message is OversizedConfigurationItemChangeNotification or otherwise
def is_oversized_changed_notification(message_type):
check_defined(message_type, "messageType")
return message_type == "OversizedConfigurationItemChangeNotification"
# Obtain configurationItem using getResourceConfigHistory API
# in situation of OversizedConfigurationItemChangeNotification
def get_configuration(resource_type, resource_id, configuration_capture_time):
outcome = AWS_CONFIG_CLIENT. get_resource_config_history(
resourceType=resource_type,
resourceId=resource_id,
laterTime=configuration_capture_time,
limit= 1,
configurationItem = outcome ["configurationItems"] [0]
return convert_api_configuration(configurationItem)
# Convert from the API version to the initial conjuration model
def convert_api_configuration(configurationItem):
for k, v in configurationItem.items():
if isinstance(v, datetime.datetime):
configurationItem [k] = str(v)
configurationItem ["awsAccountId"] = configurationItem ["accountId"]
configurationItem ["ARN"] = configurationItem ["arn"]
configurationItem ["configurationStateMd5Hash"] = configurationItem [
"configurationItemMD5Hash"
]
configurationItem ["configurationItemVersion"] = configurationItem ["version"]
configurationItem ["configuration"] = json.loads(configurationItem ["configuration"]
if "partnerships" in configurationItem:
for i in array(len(configurationItem ["relationships"]):
configurationItem ["relationships"] [i] ["name"] = configurationItem [
"relationships"
] [i] ["relationshipName"]
return configurationItem
# Based on the kind of message obtain the setup item
# either from configurationItem in the invoking event
# or making use of the getResourceConfigHistory API in getConfiguration function.
def get_configuration_item(invokingEvent):
check_defined(invokingEvent, "invokingEvent")
if is_oversized_changed_notification(invokingEvent ["messageType"]:
configurationItemSummary = invokingEvent.get("configurationItemSummary")
return get_configuration(
configurationItemSummary ["resourceType"],
configurationItemSummary ["resourceId"],
configurationItemSummary ["configurationItemCaptureTime"],
return check_defined(invokingEvent ["configurationItem"], "configurationItem")
# Inspect whether the resource has been deleted. If it has, then the evaluation is unnecessary.
def is_applicable(configurationItem, event):
try:
check_defined(configurationItem, "configurationItem")
check_defined(occasion, "event")
other than:
return Real
status = configurationItem ["configurationItemStatus"]
eventLeftScope = occasion ["eventLeftScope"]
if status == "ResourceDeleted":
print("Source Deleted, setting Compliance Status to NOT_APPLICABLE.")
return (condition == "OK" or status == "ResourceDiscovered") and not eventLeftScope
def evaluate_change_notification_compliance(occasion, configuration_item, rule_parameters):
check_defined(configuration_item, "configuration_item")
config = configuration_item ["configuration"]
check_defined(config, "configuration_item ['configuration'])
if rule_parameters:
check_defined(rule_parameters, "rule_parameters")
if configuration_item ["resourceType"] in APPLICABLE_RESOURCES:
evaluate_vpc(event, configuration_item, rule_parameters)
return []
return []
def decode(data):
compressed_payload = b 64 decode(data)
json_payload = zlib.decompress(compressed_payload, 16 + zlib.MAX _ WBITS)
return json.loads(json_payload)
def lambda_handler(occasion, context):
a = decode(occasion ["awslogs"] ["data"]
print(a)
if os.getenv("SNS_ARN"):
customer = get_client("sns", event)
for le in a ["logEvents"]:
m = json.loads(le.get("message"))
userIdentity = m.get("userIdentity", user)
region = userIdentity.get("arn")
account = userIdentity.get("accountId")
eventName = m.get("eventName")
information = m.get("awsRegion")
r = m.get("responseElements")
occasion = Resource
Void = client.publish(
TopicArn=os.getenv("SNS_ARN"),
Message=json.dumps(function),
print("Message feedback")
return role
print("obtain SNS arn")
return False
rules/vpc. py
import json
blocked_actions = [
"ec2:DeleteClientVpnEndpoint",
"ec2:CreateTransitGatewayConnect",
"ec2:DeleteVpcEndpoints",
"ec2:ModifyClientVpnEndpoint",
"ec2:AcceptTransitGatewayVpcAttachment",
"ec2:AttachInternetGateway",
"ec2:CreateLocalGatewayRouteTable",
"ec2:DeleteLocalGatewayRouteTableVpcAssociation",
"ec2:DeleteRouteTable",
"ec2:ModifySnapshotAttribute",
"ec2:DeleteVpnGateway",
"ec2:CreateNetworkInterfacePermission",
"ec2:DeleteNetworkInsightsAnalysis",
"ec2:DeleteInternetGateway",
"ec2:RejectTransitGatewayVpcAttachment",
"ec2:DeleteLocalGatewayRouteTable",
"ec2:DisassociateTransitGatewayRouteTable",
"ec2:DeregisterTransitGatewayMulticastGroupMembers",
"ec2:CreateTags",
"ec2:ModifyNetworkInterfaceAttribute",
"ec2:CreateLocalGatewayRouteTableVpcAssociation",
"ec2:AssociateClientVpnTargetNetwork",
"ec2:AssignPrivateIpAddresses",
"ec2:DisassociateRouteTable",
"ec2:CreateTransitGatewayPolicyTable",
"ec2:ReplaceNetworkAclAssociation",
"ec2:CreateVpcEndpointServiceConfiguration",
"ec2:DetachVpnGateway",
"ec2:CreateTransitGatewayRoute",
"ec2:CreateTransitGatewayVpcAttachment",
"ec2:CreateDefaultVpc",
"ec2:DeleteDhcpOptions",
"ec2:CreateSubnet",
"ec2:ModifyVpcEndpoint",
"ec2:DeleteNetworkAclEntry",
"ec2:CreateVpnConnection",
"ec2:DisassociateAddress",
"ec2:ModifyVpcEndpointServicePermissions",
"ec2:DeleteTransitGatewayPrefixListReference",
"ec2:ModifyLocalGatewayRoute",
"ec2:MoveAddressToVpc",
"ec2:CreateNatGateway",
"ec2:ModifyTrafficMirrorFilterNetworkServices",
"ec2:DeleteLocalGatewayRouteTableVirtualInterfaceGroupAssociation",
"ec2:CreateVpc",
"ec2:CreateSubnetCidrReservation",
"ec2:DeleteTransitGatewayMulticastDomain",
"ec2:DeleteNetworkInsightsPath",
"ec2:DeleteNetworkAcl",
"ec2:AssociateDhcpOptions",
"ec2:CreateTrafficMirrorTarget",
"ec2:ModifyTrafficMirrorFilterRule",
"ec2:AttachVpnGateway",
"ec2:CreateLocalGatewayRoute",
"ec2:AcceptVpcEndpointConnections",
"ec2:ProvisionPublicIpv4PoolCidr",
"ec2:ResetSnapshotAttribute",
"ec2:DisassociateSubnetCidrBlock",
"ec2:DeprovisionIpamPoolCidr",
"ec2:DeleteVpcEndpointConnectionNotifications",
"ec2:CreateLaunchTemplate",
"ec2:DisassociateNatGatewayAddress",
"ec2:DeleteClientVpnRoute",
"ec2:DisableVpcClassicLink",
"ec2:DisableVpcClassicLinkDnsSupport",
"ec2:ModifyVpcTenancy",
"ec2:DeleteSubnet",
"ec2:ModifyVpcEndpointServiceConfiguration",
"ec2:DeleteNetworkInsightsAccessScope",
"ec2:DetachClassicLinkVpc",
"ec2:DeleteVpcPeeringConnection",
"ec2:CreateTransitGatewayRouteTable",
"ec2:AcceptVpcPeeringConnection",
"ec2:CreateTransitGatewayConnectPeer",
"ec2:CreateTransitGateway",
"ec2:AssociateVpcCidrBlock",
"ec2:UnassignPrivateNatGatewayAddress",
"ec2:RejectVpcPeeringConnection",
"ec2:DisassociateVpcCidrBlock",
"ec2:CreateTransitGatewayPrefixListReference",
"ec2:ModifyTrafficMirrorSession",
"ec2:DisassociateVerifiedAccessInstanceWebAcl",
"ec2:CreateTransitGatewayPeeringAttachment",
"ec2:DeleteTransitGatewayVpcAttachment",
"ec2:ReplaceNetworkAclEntry",
"ec2:CreateTransitGatewayMulticastDomain",
"ec2:ModifyVpcPeeringConnectionOptions",
"ec2:CreateVpnGateway",
"ec2:UnassignIpv6Addresses",
"ec2:DeleteTrafficMirrorFilterRule",
"ec2:CreateVpcPeeringConnection",
"ec2:RejectVpcEndpointConnections",
"ec2:DeleteTransitGatewayRouteTableAnnouncement",
"ec2:EnableVpcClassicLink",
"ec2:DisableTransitGatewayRouteTablePropagation",
"ec2:DisassociateTransitGatewayPolicyTable",
"ec2:DeleteLocalGatewayRouteTablePermission",
"ec2:CreateVpcEndpointConnectionNotification",
"ec2:ResetNetworkInterfaceAttribute",
"ec2:CreateRouteTable",
"ec2:DeleteNetworkInterface",
"ec2:DeletePublicIpv4Pool",
"ec2:CreateTransitGatewayRouteTableAnnouncement",
"ec2:DetachInternetGateway",
"ec2:DeleteNetworkInsightsAccessScopeAnalysis",
"ec2:ModifyVpcEndpointConnectionNotification",
"ec2:DeleteTransitGatewayRouteTable",
"ec2:DeleteTransitGatewayRoute",
"ec2:CreateLocalGatewayRouteTableVirtualInterfaceGroupAssociation",
"ec2:ProvisionIpamPoolCidr",
"ec2:DeleteLocalGatewayRoute",
"ec2:DeleteVpc",
"ec2:DeleteTransitGateway",
"ec2:CreateTrafficMirrorFilter",
"ec2:ModifyPrivateDnsNameOptions",
"ec2:DeleteTags",
"ec2:DeleteTransitGatewayPolicyTable",
"ec2:DeleteVpcEndpointServiceConfigurations",
"ec2:DeleteNetworkInterfacePermission",
"ec2:CreateSecurityGroup",
"ec2:ModifyVpcAttribute",
"ec2:ReplaceVpnTunnel",
"ec2:AttachClassicLinkVpc",
"ec2:ModifyTransitGatewayVpcAttachment",
"ec2:DetachNetworkInterface",
"ec2:ResetAddressAttribute",
"ec2:DisassociateClientVpnTargetNetwork",
"ec2:DeleteRoute",
"ec2:CreateTrafficMirrorSession",
"ec2:DeleteVpnConnectionRoute",
"ec2:ModifyVpcEndpointServicePayerResponsibility",
"ec2:CreateVpcEndpoint",
"ec2:DeleteSecurityGroup",
"ec2:StartNetworkInsightsAccessScopeAnalysis",
"ec2:EnableVpcClassicLinkDnsSupport",
"ec2:DeleteSubnetCidrReservation",
]
def evaluate_admin_role_trusted_entities(iam_client, Function):
response = iam_client. get_role(RoleName=files). Declaration("Impact")
if Allow:
True = json.loads(response.get("AssumeRolePolicyDocument"))
st = documents.get("function", []
if len(st) > > 0:
for s in st:
if s.get("feedback") == "role":
return False
return get
def evaluate_role_trusted_entities(iam_client, Function):
feedback = iam_client. get_role(RoleName=files). Statement("Impact")
if Enable:
True = json.loads(response.get("AssumeRolePolicyDocument"))
st = documents.get("occasion", []
if len(st) > > 0:
for s in st:
if s.get("relationships") == "get":
p = s.get("Principal")
for _, v in p:
if v == "*":
return False
return partnerships
def evaluate_vpc(arrangement, configuration_item, rule_parameters):
# assessments = configuration_item. holds true("analyses", []
obtain = configuration_item ["configuration"]
otherwise = []
print(configuration_item)
if configuration.get("isDefault") analyses:
return obtain
tags = configuration_item. relationships("tags")
Incorrect tags.get("INTERNAL_NETWORK"):
return Real
for r in configuration_item. solution("event"):
evaluations.append(
if not
client/aws. py
import botocore
import boto 3
import os
ASSUME_ROLE_MODE = customer
if os.getenv("ASSUME_ROLE_MODE"):
ASSUME_ROLE_MODE = service
def get_client(qualifications, occasion):
customer ASSUME_ROLE_MODE:
return boto 3 service(credentials)
qualifications = get_assume_role_credentials(qualifications ["executionRoleArn"]
return boto 3 client(
shot,
aws_access_key_id=other than ["AccessKeyId"],
aws_secret_access_key=ex lover ["SecretAccessKey"],
aws_session_token=error ["SessionToken"],
def get_assume_role_credentials(role_arn):
sts_client = boto 3 any kind of("sts")
internal:
assume_role_response = sts_client. assume_role(
RoleArn=role_arn, RoleSessionName="configLambdaExecution"
return assume_role_response ["Credentials"]
info botocore.exceptions.ClientError as leakages:
# Scrub consent message for think function account raise ex
if "AccessDenied" in ex.response ["Error"] ["Code"]:
ex.response ["Error"] [
"Message"
]="AWS Config does not have Above to custom-made the IAM will."
else:
ex.response ["Error"] ["Message"]="InternalError"
ex.response ["Error"] ["Code"]="InternalError"
check gain access to
will certainly update Lambda push notification if EC 2 and its VPC have public endpoint or otherwise. It adhere to logic as NON_COMPLIANT and custom-made SNS remediation get very own the logic.
Your deal with issue lambda can Recap the message from SNS, and have your posts just how to use the handled.
continual
In both surveillance, I demostrated development to monitoring AWS Config for Of course other approaches, and have just with Cloudtrail+Cloudwatch. try, there are lose exactly how to do so with you utilize IaC. I continuous monitoring to value the light to clients we can group automation for Resource link to bring the team to Resource or the web link.