AWS CloudFront generate signed urls and cookies with Python and Golang

documents

create key group

generate key pair

openssl genrsa -out private_key.pem 2048

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

upload public key and create key group

• upload public key: CloudFront -> Public keys -> Add public key
• create key group: CloudFront -> Key groups -> Add key group

Enable Restrict Viewer Access (Use Signed URLs or Signed Cookies)

CloudFront -> Distributions -> ** your distribution ** -> Behaviors -> Edit

• Restrict Viewer Access (Use Signed URLs or Signed Cookies): choose Yes
• Trusted Key Groups or Trusted Signer: choose Trusted Key Groups
• Trusted Key Groups: choose your created key group

Python code

"""
generate signed urls or cookies for AWS CloudFront

pip install botocore rsa requests
"""
from datetime import datetime, timedelta
import functools
from urllib.parse import urlsplit

from botocore.signers import CloudFrontSigner
import requests
import rsa


class CloudFrontUtil:
    def __init__(self, private_key_path: str, key_id: str):
        """
        :param private_key_path: str, the path of private key which generated by openssl command line
        :param key_id: str, CloudFront -> Key management -> Public keys
        """
        self.key_id = key_id

        with open(private_key_path, 'rb') as fp:
            priv_key = rsa.PrivateKey.load_pkcs1(fp.read())

        # NOTE: CloudFront use RSA-SHA1 for signing URLs or cookies
        self.rsa_signer = functools.partial(
            rsa.sign, priv_key=priv_key, hash_method='SHA-1'
        )
        self.cf_signer = CloudFrontSigner(key_id, self.rsa_signer)

    def generate_presigned_url(self, url: str, expire_at: datetime) -> str:
        # Create a signed url that will be valid until the specfic expiry date
        # provided using a canned policy.
        return self.cf_signer.generate_presigned_url(url, date_less_than=expire_at)

    def generate_signed_cookies(self, url: str, expire_at: datetime) -> str:
        policy = self.cf_signer.build_policy(url, expire_at).encode('utf8')
        policy_64 = self.cf_signer._url_b64encode(policy).decode('utf8')

        signature = self.rsa_signer(policy)
        signature_64 = self.cf_signer._url_b64encode(signature).decode('utf8')
        return {
            "CloudFront-Policy": policy_64,
            "CloudFront-Signature": signature_64,
            "CloudFront-Key-Pair-Id": self.key_id,
        }


if __name__ == '__main__':
    private_key_path = './private_key.pem'  # generated by openssl command
    key_id = 'xxxxx'  # CloudFront -> Key management -> Public keys, the value of `ID` field
    url = 'https://xxxxx.cloudfront.net/project-abc/README.md'  # your file's cdn url
    expire_at = datetime.now() + timedelta(days=1)

    cfu = CloudFrontUtil(private_key_path, key_id)

    obj_key = urlsplit(url).path
    
    # signed cookies
    signed_cookies = cfu.generate_signed_cookies(url, expire_at)
    r = requests.get(url, cookies=signed_cookies)
    print(f'using signed cookie: {obj_key}, {r.status_code}, {r.content}')

    # signed url
    signed_url = cfu.generate_presigned_url(url, expire_at)
    r = requests.get(signed_url)
    print(f'\nusing signed url: {obj_key}, {r.status_code}, {r.content}')

golang code

package main

import (
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"time"

	"github.com/aws/aws-sdk-go-v2/service/cloudfront/sign"
)

func main() {
	keyID := "XXXXX"  // CloudFront -> Key management -> Public keys
	privKeyPath := "./private_key.pem"  // generated by openssl command line
	url := "https://xxxxx.cloudfront.net/abc/README.md"  // change it for yourself
	expireAt := time.Now().Add(24 * time.Hour)
    
	privKey, err := sign.LoadPEMPrivKeyFile(privKeyPath)
	if err != nil {
		log.Fatalf("Load private key from %s failed\n", privKeyPath)
	}

	URLSigner := sign.NewURLSigner(keyID, privKey)

	// generate signed url
	signedURL, err := URLSigner.Sign(url, expireAt)
	if err != nil {
		log.Fatal("generate signed url failed:", err)
	}
	fmt.Printf("access signedURL: %s\ncontent: %s", signedURL, httpGet(signedURL))

	// generate signed cookies
	cookieSigner := sign.NewCookieSigner(keyID, privKey)
	signedCookies, err := cookieSigner.Sign(url, expireAt)
	if err != nil {
		log.Fatal("generate signed cookie failed:", err)
	}
	fmt.Printf("access with signedCookies: %s\ncontent: %s", signedCookies, httpGetWithCookie(url, signedCookies))
}

func httpGet(url string) string {
	res, err := http.Get(url)

	if err != nil {
		log.Fatal(err)
	}
	defer res.Body.Close()

	data, _ := ioutil.ReadAll(res.Body)

	return fmt.Sprintf("%s", data)
}

func httpGetWithCookie(url string, cookies []*http.Cookie) string {
	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		log.Fatal(err)
	}

	for _, c := range cookies {
		req.AddCookie(c)
	}

	client := &http.Client{}
	res, err := client.Do(req)

	if err != nil {
		log.Fatal(err)
	}
	defer res.Body.Close()

	data, _ := ioutil.ReadAll(res.Body)

	return fmt.Sprintf("%s", data)
}