<AWS> Docker Container에서 Boto3로 IAM role 자격 증명 얻어서 서비스 접근하기

이번 블로그에서는

access key, secret key를 IAM role로 전환
elasticsearch 접근을 위해 AWS4Auth로 얻은 자격 증명을 elasticsearch에 넘겨줄 때 발생하는 403 The security token included in the request is invalid 오류 해결
AWS 서비스 사용할 때 session token을 확인하고 필요하면 refresh하기

에 대해서 알게된 점들을 정리하였다.

본문에서는 IAM role, Elasticsearch 등 AWS 서비스가 나올텐데 이에 대한 설명은 이 블로그에서는 진행하지 않을 예정이며 기초 지식을 가졌다는 가정하에 과정을 설명하도록 할 것임을 미리 밝혀둔다.

About

필자의 회사에서는 서비스를 론칭하기에 앞서 개발하면서 서로 AWS access key, secret key를 환경 변수로 주입하는 등의 방법을 주로 사용했었는데, 사실상 모든 혹은 대부분의 자원을 AWS에서 운영 중인 우리의 경우에는 그보다 더 안전하게 IAM role을 이용할 수 있었기 때문이다. 그래서 django-environ를 이용해서 Key에 접근하던 방식을 버리고 boto3, AWS4Auth로 전환하였다.

사용한 AWS 자원 및 python 패키지는 아래와 같다.

AWS
- EC2
- Elasticsearch
- S3
Python package
- boto3
- requests-aws4auth

(참고로 ECS fargate에서도 같은 방법을 사용했을 때 문제없이 진행되는 것은 확인했다.)

Implementation

get_credentials()

import boto3

credentials = boto3.Session().get_credentials()

get_credentials()은 credential을 얻어오는 메써드다. boto3가 자격 증명을 얻어오는 구조를 살펴보면 우선순위들이 있다. 이와 관련해서는 블로그를 보면 아주 상세히 설명돼있으니 확인하시면 되겠다.

필자는 2가지 경우에 credentials가 가지는 값이 어떻게 다른지 확인을 먼저 해보았다.

1. aws cli를 통해서 configure한 경우

즉, access key, secret key를 ~/.aws/config에서 가져오는 경우이다. credentials.__dict__로 값을 찍어보면,

# AWS CONFIGURE
{
    'access_key': 'ABCDEFGHIJKLMNOPQRSTUV',
    'secret_key': 'ABCDEFGHIJKLMNOPQRSTUV123456789',
    'token': None,
    'method': 'shared-credentials-file'
}

처럼 나온다. token은 따로 부여하지 않는다는 점을 확인할 수 있다.

2. EC2에서 IAM role로부터 가져온 경우

# IAM ROLE
{
    '_refresh_using': <bound method InstanceMetadataFetcher.retrieve_iam_role_credentials of <botocore.utils.InstanceMetadataFetcher object at 0x7fb865cd64e0>>,
    '_access_key': 'ABCDEFGHIJKLMNOPQRSTUV', 
    '_secret_key': 'ABCDEFGHIJKLMNOPQRSTUV123456789', 
    '_token': '<token>',
    '_expiry_time': datetime.datetime(2020, 2, 19, 17, 16, 6, tzinfo=tzlocal()), 
    '_time_fetcher': <function _local_now at 0x7fb8667cdea0>, 
    '_refresh_lock': <unlocked _thread.lock object at 0x7fb86b270ad0>, 
    'method': 'iam-role', # ECS fargate 경우에는 container-role로 뜬다.
    '_frozen_credentials': ReadOnlyCredentials(
        access_key='ABCDEFGHIJKLMNOPQRSTUV', 
        secret_key='ABCDEFGHIJKLMNOPQRSTUV123456789', 
        token='<token>',
    )
}

참고로 token값은 너무 길어서 그냥 <token>으로 넣어둔 것이다. 중요한 점은 이 경우에는 token이 존재한다는 점이다. 이 두 경우의 차이 때문에 로컬이거나 aws key들이 configure 돼 있는 기기에서는 발생하지 않던 오류가 IAM role을 통해서 접근하려할 때면 403 The security token included in the request is invalid 오류를 던지게 되는 것이다.

해결 방법은 단순하다. 그냥 session token을 넣어주기만 하면 된다.

import boto3
service_name = 's3'
region_name = 'ap-northeast-2'
credentials = boto3.Session().get_credentials()
boto3.client(
    service_name,
    aws_access_key_id=credentials.access_key,
    aws_secret_access_key=credentials.secret_key,
    region_name=region_name,
    aws_session_token=credentials.token,
)

필자의 기억이 맞다면 AWS4Auth로 자격 증명을 Token없이 얻고 Elasticsearch 객체를 만들 때, 403 The security token included in the request is invalid 오류가 발생했을 것이다. 그 경우도 역시 아래처럼 session_token을 넣어주는 방식으로 해결할 수 있다.

import boto3
from requests_aws4auth import AWS4Auth
from elasticsearch import Elasticsearch, RequestsHttpConnection

service_name = 'es'
region_name = 'ap-northeast-2'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(
    credentials.access_key,
    credentials.secret_key,
    region_name,
    service_name,
    session_token=credentials.token,
)

Elasticsearch(
    hosts='AWS_ELASTIC_HOST',
    port=9200,
    http_auth=awsauth,
    use_ssl=True,
    verify_certs=True,
    connection_class=RequestsHttpConnection
)

위의 코드로 접근 권한을 얻는 것은 해결했다. 이제 문제는 서비스에서 es에 접근하는 코드 혹은 다른 서비스에 접근하는 코드에서 token이 유효한지 확인 후에 유효하지 않으면 새로운 자격 증명을 받아서 해당 서비스를 접근하는 코드만 구현하면 된다.

필자는 글로벌하게 접근할 때 하나의 aws객체만 사용할 수 있도록 singleton으로 클래스를 만들어서 구현했다. AWS 서비스는 @property로 지정하여, 매번 접근할 때면 token이 유효한지 확인하고 만료된 경우에는 재발급해서 보내주는 방식을 사용했다.

import boto3
import datetime
from requests_aws4auth import AWS4Auth
from elasticsearch import Elasticsearch, RequestsHttpConnection

ELASTIC_HOST = 'elasticsearch-url'
AWS_REGION = 'ap-northeast-2'


def singleton(_cls):
    instances = {}

    def getinstance():
        if _cls not in instances:
            instances[_cls] = _cls()
        return instances[_cls]

    return getinstance


@singleton
class AWS():

    def __init__(self):
        self._set_credentials()
        self._set_services()

    def _set_services(self):
        self.__es = self.get_es()
        self.__s3 = self.get_service_client_by_name('s3')

    def _set_credentials(self):
        self.__credentials = boto3.Session().get_credentials()

        # shared-credentials-file 처럼 임시 자격 증명이 아닌 경우에는 
        # _expiry_time 이 항상 존재한다.
        if '_expiry_time' in self.__credentials.__dict__:
            self.__expiry_time = self.__credentials._expiry_time
        else:
            self.__expiry_time = None

    def _refresh(self):
        if self._is_token_expired():
            print('Refreshed aws session token')
            self._set_credentials()
            self._set_services()
    
    def _is_token_expired(self):
        if self.__expiry_time is None:
            return False

        tzinfo = self.__expiry_time.tzinfo
        return self.__expiry_time < datetime.datetime.now(tzinfo)

    def get_es(self):
        awsauth = AWS4Auth(
            self.__credentials.access_key,
            self.__credentials.secret_key,
            AWS_REGION,
            'es',
            session_token=self.__credentials.token,
        )

        return Elasticsearch(
            hosts=ELASTIC_HOST,
            port=9200,
            http_auth=awsauth,
            use_ssl=True,
            verify_certs=True,
            connection_class=RequestsHttpConnection
        )

    def get_service_client_by_name(self, service_name):
        return boto3.client(
            service_name,
            region_name=AWS_REGION,
        )

    @property
    def es(self):
        self._refresh()
        return self.__es

    @property
    def s3(self):
        self._refresh()
        return self.__s3

Conculusion

간단하게 boto3를 이용하여 Token을 refresh하는 방법을 알아보았다. 하지만, 실제 프로덕트에서 만약 aws service에 대한 접근이 정말 많다면 매번 self._refresh()를 실행하는 것이 현명하지는 않을 것임을 명심하자. 그런 경우에는 Auto-refresh AWS Tokens Using IAM Role and boto3 블로그에서 소개하는 방식을 사용하는 것이 더 좋아보이는데, 아직 필자는 테스트해보지 않아서 따로 소개하지는 않았지만, 의문이 드시는 분들을 위해 링크를 소개해드리면서 글을 마치고자 한다.

참고로 Dockerize 후에 Docker container에서도 위의 코드들은 똑같이 작동한다. 위에서도 이미 언급한 바 있지만, ECS fargate에서도 잘 작동하는 것을 확인했으니 필요하신 분들은 참고하시길 바란다.