[AWS] AWS Config로 Security group 감시하기

AWS Config는 AWS, 온프레미스 및 기타 클라우드에서 리소스의 구성과 관계를 지속적으로 진단, 감사 및 평가합니다. 리소스를 지속적으로 감시하여 상태가 변하지 않고 유지되게 만들 수 있습니다. 오늘은 security group을 감시하여 inbound rule이 추가되거나 제거되지 않게 해 봅시다.

1. Config를 제외한 리소스 준비

inbound rule로 22번(SSH) 포트만 허용합니다. 인스턴스로 SSH 연결만 한다는 가정으로 ssh inbound rule만 유지되게 만드는 것입니다.

Python 런타임을 사용하여 Lambda Function을 생성합니다. 다음 글을 참고하여 Lambda 코드를 작성합니다.

https://aws.amazon.com/ko/blogs/security/how-to-monitor-aws-account-configuration-changes-and-api-calls-to-amazon-ec2-security-groups/

How to Monitor AWS Account Configuration Changes and API Calls to Amazon EC2 Security Groups | Amazon Web Services

REQUIRED_PERMISSIONS에 없는 inbound rule이 있을 경우 삭제(revoke)하고, REQUIRED_PERMISSIONS에 있지만 inbound rule에 없는 경우 추가(authorize)합니다. 또한, 리소스의 상태를 반환합니다.

• NOT_APPLICABLE : AWS Config가 보안그룹이 아닌 리소스를 감시하고 있거나, 감시하고 있는 리소스가 삭제되었습니다.
• NON_COMPLIANT : 보안 그룹이 규칙 준수 조건에 부합하지 않는 상태입니다.
• COMPLIANT : 보안 그룹이 규칙 준수 조건에 부합하는 상태입니다.

import boto3
import botocore
import json

APPLICABLE_RESOURCES = ["AWS::EC2::SecurityGroup"]

REQUIRED_PERMISSIONS = [
{
    "IpProtocol" : "tcp",
    "FromPort" : 22,
    "ToPort" : 22,
    "UserIdGroupPairs" : [],
    "IpRanges" : [{"CidrIp" : "0.0.0.0/0"}],
    "PrefixListIds" : [],
    "Ipv6Ranges" : []
}]

def normalize_parameters(rule_parameters):
    for key, value in rule_parameters.items():
        normalized_key=key.lower()
        normalized_value=value.lower()

        if normalized_value == "true":
            rule_parameters[normalized_key] = True
        elif normalized_value == "false":
            rule_parameters[normalized_key] = False
        elif normalized_value.isdigit():
            rule_parameters[normalized_key] = int(normalized_value)
        else:
            rule_parameters[normalized_key] = True
    return rule_parameters

def evaluate_compliance(configuration_item, debug_enabled):
    if configuration_item["resourceType"] not in APPLICABLE_RESOURCES:
        return {
            "compliance_type" : "NOT_APPLICABLE",
            "annotation" : "The rule doesn't apply to resources of type " +
            configuration_item["resourceType"] + "."
        }

    if configuration_item["configurationItemStatus"] == "ResourceDeleted":
        return {
            "compliance_type": "NOT_APPLICABLE",
            "annotation": "The configurationItem was deleted and therefore cannot be validated."
        }

    group_id = configuration_item["configuration"]["groupId"]
    client = boto3.client("ec2");

    try:
        response = client.describe_security_groups(GroupIds=[group_id])
    except botocore.exceptions.ClientError as e:
        return {
            "compliance_type" : "NON_COMPLIANT",
            "annotation" : "describe_security_groups failure on group " + group_id
        }
        
    if debug_enabled:
        print("security group definition: ", json.dumps(response, indent=2))

    ip_permissions = response["SecurityGroups"][0]["IpPermissions"]
    authorize_permissions = [item for item in REQUIRED_PERMISSIONS if item not in ip_permissions]
    revoke_permissions = [item for item in ip_permissions if item not in REQUIRED_PERMISSIONS]

    if authorize_permissions or revoke_permissions:
        annotation_message = "Permissions were modified."
    else:
        annotation_message = "Permissions are correct."

    if authorize_permissions:
        if debug_enabled:
            print("authorizing for ", group_id, ", ip_permissions ", json.dumps(authorize_permissions, indent=2))

        try:
            client.authorize_security_group_ingress(GroupId=group_id, IpPermissions=authorize_permissions)
            annotation_message += " " + str(len(authorize_permissions)) +" new authorization(s)."
        except botocore.exceptions.ClientError as e:
            return {
                "compliance_type" : "NON_COMPLIANT",
                "annotation" : "authorize_security_group_ingress failure on group " + group_id
            }

    if revoke_permissions:
        if debug_enabled:
            print("revoking for ", group_id, ", ip_permissions ", json.dumps(revoke_permissions, indent=2))

        try:
            client.revoke_security_group_ingress(GroupId=group_id, IpPermissions=revoke_permissions)
            annotation_message += " " + str(len(revoke_permissions)) +" new revocation(s)."
        except botocore.exceptions.ClientError as e:
            return {
                "compliance_type" : "NON_COMPLIANT",
                "annotation" : "revoke_security_group_ingress failure on group " + group_id
            }

    return {
        "compliance_type": "COMPLIANT",
        "annotation": annotation_message
    }

def lambda_handler(event, context):
    invoking_event = json.loads(event['invokingEvent'])
    configuration_item = invoking_event["configurationItem"]
    rule_parameters = normalize_parameters(json.loads(event["ruleParameters"]))

    debug_enabled = False

    if "debug" in rule_parameters:
        debug_enabled = rule_parameters["debug"] 

    if debug_enabled:
        print("Received event: " + json.dumps(event, indent=2))

    evaluation = evaluate_compliance(configuration_item, debug_enabled)

    config = boto3.client('config')

    response = config.put_evaluations(
       Evaluations=[
           {
               'ComplianceResourceType': invoking_event['configurationItem']['resourceType'],
               'ComplianceResourceId': invoking_event['configurationItem']['resourceId'],
               'ComplianceType': evaluation["compliance_type"],
               "Annotation": evaluation["annotation"],
               'OrderingTimestamp': invoking_event['configurationItem']['configurationItemCaptureTime']
           },
       ],
       ResultToken=event['resultToken'])

Lambda가 AWS Config와 보안그룹에 접근할 수 있도록 Lambda IAM Role에 Policy를 만들어 추가해 줍니다.

{
	"Version": "2012-10-17",
	"Statement": [
		{
            "Effect": "Allow",
            "Action": [
                "config:PutEvaluations",
                "ec2:DescribeSecurityGroups",
                "ec2:AuthorizeSecurityGroupIngress",
                "ec2:RevokeSecurityGroupIngress"
            ],
            "Resource": "*"
        }
	]
}

Lambda가 시간 부족으로 실패하지 않게 Timeout을 1분으로 바꿔줍니다.

2. AWS Config 시작

Config 콘솔에서 Get started를 누릅니다.

지속적으로 EC2 SecurityGroup을 감시하도록 설정합니다. 다른 설정은 건드리지 않고 Next를 누릅니다.

AWS Managed Rule에 보안그룹을 관리하는 규칙이 없기 때문에 스킵하고 Confirm을 누릅니다. 이제 AWS Config에서 보안그룹을 지속적으로 감시할 수 있으며, 그에 관한 규칙을 생성할 수 있습니다.

이제 규칙을 생성하기 위해 Rules에서 Add rule을 누릅니다.

Lambda를 사용하여 리소스의 상태를 정의하고 관리하기 위해 Create custom Lambda rule을 선택합니다. Next를 누릅니다.

규칙의 이름과 Lambda Function의 Arn을 입력합니다.

프로비저닝 된 리소스만 평가하게 하고, 특정한 보안그룹의 설정이나 값이 변경되었을 때만 Lambda Function이 평가하도록 합니다.

Next를 누른 뒤 Save를 눌러 규칙을 생성합니다.

3. 테스트

inbound rule에 ssh rule에 더해 http rule도 추가해 줍니다.

Save rules를 누른 뒤, 약 1분 정도를 기다리면 http rule이 삭제되고, ssh rule만 남은 것을 볼 수 있습니다.

오늘의 글은 여기까지입니다. 감사합니다!