Cloudfront에 signed cookie로 access 하기

Cloudfront에 signed cookie로 access 하기

Signed cookie를 이용하게 된 배경

Cloudfront에서 특정 유저만 url에 접근하도록 하고 싶은 경우 사용할 수 있는 방법은 크게 3가지가 있다.

• pre-signed url
• signed url
• signed cookie

만약 프론트엔드가 url에 접근하려 한다면 signed url과 pre-signed url을 이용하는 방법이 있는데 pre-signed url을 이용할 경우 cognito를 이용해야 한다는 번거로움이 있다. 또한, S3의 특정 디렉토리에 있는 여러 개의 파일을 가져와야 하는데 이를 위해서는 signed cookie가 필요하다. 우리는 여러 개의 파일에 접근해야 하기 때문에 백엔드에서 signed cookie를 받아와 이를 프론트엔드에 전달해 주는 작업을 하려고 한다. 과정은 다음과 같다.

목차

AWS 인프라 파트

1. S3 버킷 생성
2. 암호화 작업
   • pem key 다운로드(퍼블릭 키 생성)
   • Cloudfront에 public key 추가하기
     - key groups 생성
3. Cloudfront 배포 생성 전
   • 원본 Access ID 생성
4. Cloudfront 배포 생성
   • S3를 origin으로 해야 함
   • 위에서 생성한 key groups 연결
   • 원본 Access ID에 대한 OAI 생성

소스 코드 파트

1. signed cookie를 가져오기 위한 코드 작업
2. 테스트

AWS 인프라 작업

S3 버킷 생성

• 퍼블릭 액세스를 차단한 버킷을 새로 생성한다.

암호화 작업

pem key 다운로드

다음과 같은 방법으로 public key와 private key를 다운 받아 key pair를 생성할 수 있다(aws 공식 문서 참고).

• openssl을 이용해 2048비트의 RSA 키 페어를 생성하고 private_key.pem 파일에 저장한다.
  - 이렇게 만들어진 파일은 퍼블릭 키와 프라이빗 키를 모두 포함하고 있다.

	openssl genrsa -out private_key.pem 2048

• 이제 private_key.pem 파일에서 public_key.pem 파일로 공개키를 추출한다.

	openssl rsa -pubout -in private_key.pem -out public_key.pem

Cloudfront에 public key 추가하기

• CloudFront > Public key > 퍼블릭 키 생성 에서 public key를 등록한다.
  - 식별하기 쉬운 Name을 지정한 후(Description도 넣으면 좋다) Key에 cat public_key.pem으로 나온 내용을 복붙한다.

Cloudfront에 key groups 등록하기

• 원하는 이름을 추가한다.
• public key에 위에서 생성한 public key를 선택한다.
  - 1 ~ 5개까지의 key를 한 그룹에 등록할 수 있다.

Cloudfront 배포 생성 전 작업

• 배포를 생성하기 전에 원본 액세스 ID를 가지고 있어야 한다.
• 이는 S3에 액세스 하기 위해 Cloudfront에 지정할 액세스 ID이다.
  - 이를 S3 OAI 생성시 ID로 사용한다.

Cloudfront 배포

• 오리진은 위에서 생성한 S3로 선택한다.
• 주의해야 할 점은 OAI를 사용하도록 설정해야 한다는 것이다.
  - 배포 생성 전 만든 원본 액세스 ID를 선택하고, 버킷 정책 업데이트를 선택하면 Cloudfront 배포 생성시 S3에서 정책이 알아서 업데이트 된다.
  

소스코드 작업

boto3 패키지에서 signed cookie 생성은 지원이 안된다.... 그렇기 때문에 직접 코드를 짜야 한다. 그러나 여러 사람들이 짜 놓은 코드들이 있었는데 이를 가져와서 조금의 변형을 거쳐 코드를 완성했다. 참고한 소스 코드 링크는 여기 있다.

수정한 코드

• signed cookies 생성에 사용되는 함수들을 클래스화 하여 어떤 함수를 호출해야 하는지 조금 더 명확히 하고자 했다.
  - 나중에 사용하려고 할 때 어떤 함수를 호출해야 하는지 헷갈릴 것 같아 클래스화 했다.
  - 또한, 타입 어노테이션을 추가해 파라미터 타입과 리턴 타입을 명시했다.
• 실제로 signed cookies를 생성할 때는 get_signed_cookies 함수만 호출하면 된다.
  - signed cookies는 세 가지로 구성 된다.
  - CloudFront-Policy, CloudFront-Signature, CloudFront-Key-Pair-Id
• 여기서 사용된 cloudfront id는 cloudfront의 public key id이다.
  - 위에서 Cloudfront에 public key를 추가할 때 생성된 public key id이다. 이 부분이 헷갈릴 것 같아 변수명을 조금 더 정확히 명시했다.
  

class SignedCookieGenerator:
    """
    클라이언트에서 cloudfront로부터 원하는 이미지를 가져오기 위해 필요한 signed cookie를 생성한다.
    """

    @staticmethod
    def _replace_unsupported_chars(some_str: str) -> str:
        """Replace unsupported chars: '+=/' with '-_~'"""
        return some_str.replace("+", "-").replace("=", "_").replace("/", "~")

    @staticmethod
    def _in_an_hour() -> int:
        """Returns a UTC POSIX timestamp for one hour in the future"""
        return int(time.time()) + (60 * 60)

    @staticmethod
    def rsa_signer(message: bytes, key: bytes) -> bytes:
        """
        Updated version
        Based on https://boto3.readthedocs.io/en/latest/reference/services/cloudfront.html#examples
        """
        private_key = serialization.load_pem_private_key(
            key, password=None, backend=default_backend()
        )
        signature = private_key.sign(message, padding.PKCS1v15(), hashes.SHA1())
        return signature

    def generate_policy_cookie(self, url: int) -> tuple:
        """Returns a tuple: (policy json, policy base64)"""

        # expires in an hour
        policy_dict = {
            "Statement": [
                {
                    "Resource": url,
                    "Condition": {
                        "DateLessThan": {"AWS:EpochTime": self._in_an_hour()}
                    },
                }
            ]
        }

        # Using separators=(',', ':') removes seperator whitespace
        policy_json = json.dumps(policy_dict, separators=(",", ":"))

        policy_64 = str(base64.b64encode(policy_json.encode("utf-8")), "utf-8")
        policy_64 = self._replace_unsupported_chars(policy_64)
        return policy_json, policy_64

    def generate_signature(self, policy: json, private_key: bytes):
        """Creates a signature for the policy from the key, returning a string"""
        sig_bytes = self.rsa_signer(policy.encode("utf-8"), private_key)
        sig_64 = self._replace_unsupported_chars(
            str(base64.b64encode(sig_bytes), "utf-8")
        )
        return sig_64

    @staticmethod
    def generate_cookies(policy: json, signature: str, cloudfront_id: str) -> dict:
        """Returns a dictionary for cookie values in the form 'COOKIE NAME': 'COOKIE VALUE'"""
        return {
            "CloudFront-Policy": policy,
            "CloudFront-Signature": signature,
            "CloudFront-Key-Pair-Id": cloudfront_id,
        }

    def generate_signed_cookies(
        self, url: int, cloudfront_id: str, private_key: bytes
    ) -> dict:
        policy_json, policy_64 = self.generate_policy_cookie(url)
        signature = self.generate_signature(policy_json, private_key)
        return self.generate_cookies(policy_64, signature, cloudfront_id)


def get_signed_cookies() -> dict:
    """
    How to get pem keys
    1. private_key with RSA
        $ openssl genrsa -out private_key.pem 2048
    2. public_key from private_key
        $ openssl rsa -pubout -in private_key.pem -out public_key.pem

    Signed Cookie consists of 3 keys
    """

    CLOUDFRONT_URL = settings.CLOUDFRONT_URL
    CLOUDFRONT_PUBLIC_KEY_ID = os.getenv("CLOUDFRONT_PUBLIC_KEY_ID")

    signed_cookie_generator = SignedCookieGenerator()
    path_to_pem_key = os.path.join(settings.BASE_DIR, "private_key.pem")

    with open(path_to_pem_key, "rb") as f:
        cookies = signed_cookie_generator.generate_signed_cookies(
            url=CLOUDFRONT_URL,
            cloudfront_id=CLOUDFRONT_PUBLIC_KEY_ID,
            private_key=f.read(),
        )

    return cookies

테스트

프론트엔드에 signed cookie를 전달해 주기 전에 실제로 원하는 url에 접근하려 할 때 signed cookie를 이용할 수 있는 지를 테스트 해 보았다.
테스트 코드는 다음과 같다.

import requests

from utils.aws import get_signed_cookies  # 위에서 생성한 get_signed cookies 함수 임포트(utils/aws.py 파일에 생성함)


def images_request_test() -> None:
    """
    signed cookie를 이용해 cloudfront에 요청시 status code 200이 오는지 테스트 하기 위한 스크립트
    문제 발생시를 대비해 임시로 사용할 예정
    """

    cloudfront_url = CLOUDFRONT_URL
    cookies = get_signed_cookies()
    print("signed cookie 정보", cookies)

    images = [
        image_id_1, image_id_2, ...
    ]
    try:
        print("요청을 시작합니다.")
        responses = []
        for image in images:
            key = f"test/{image}.png"
            url = f"{cloudfront_url}/{key}"
            response = requests.get(url, cookies=cookies)
            print(response.status_code)
            responses.append(response.status_code)
        print(responses)
    except Exception as e:
        print(e)
    print("끄읕")