Streaming and Delivery on AWS

import base64
import datetime

from botocore.signers import CloudFrontSigner
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding

KEY_PAIR_ID = "K2JCJMDEHXQW5F"               # the public key ID in CloudFront
with open("private_key.pem", "rb") as f:
    PRIVATE_KEY = serialization.load_pem_private_key(f.read(), password=None)


def _sign(message: bytes) -> bytes:
    return PRIVATE_KEY.sign(message, padding.PKCS1v15(), hashes.SHA1())    # CloudFront requires RSA SHA-1


def _cloudfront_b64(data: bytes) -> str:
    return base64.b64encode(data).replace(b"+", b"-").replace(b"=", b"_").replace(b"/", b"~").decode()


def signed_cookies(media_id: str, hours: int = 2) -> dict[str, str]:
    expires = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=hours)
    signer = CloudFrontSigner(KEY_PAIR_ID, _sign)
    policy = signer.build_policy(f"https://cdn.example.com/videos/{media_id}/*", expires).encode()
    return {
        "CloudFront-Policy": _cloudfront_b64(policy),
        "CloudFront-Signature": _cloudfront_b64(_sign(policy)),
        "CloudFront-Key-Pair-Id": KEY_PAIR_ID,
    }

import { getSignedCookies } from '@aws-sdk/cloudfront-signer';

export function signedCookies(mediaId, privateKey, keyPairId) {
  const url = `https://cdn.example.com/videos/${mediaId}/*`;
  const policy = JSON.stringify({
    Statement: [{ Resource: url, Condition: { DateLessThan: { 'AWS:EpochTime': Math.floor(Date.now() / 1000) + 7200 } } }],
  });
  return getSignedCookies({ keyPairId, privateKey, policy });
  // { 'CloudFront-Policy': '...', 'CloudFront-Signature': '...', 'CloudFront-Key-Pair-Id': '...' }
}