Подписанные (pre-signed) URL для Яндекс Облака

readme.md

Подписанные (pre-signed) URL

Реализация алгоритма создания подписанных запросов (https://cloud.yandex.ru/docs/storage/concepts/pre-signed-urls) на PHP и Python3.

yc_urls.php

<?php 

  // Usage:
  // include('yc-urls.php');
  // $ycu = new YandexCloudUrls('JK38EXAMPLEAKDID8', 'ExamP1eSecReTKeykdokKK38800');
  // $time = DateTime::createFromFormat('Y-m-d H:i:s', '2019-08-01 00:00:00');
  // echo $ycu->encodeUrl('/example-bucket/object-for-share.txt', $time, 60*60);
  class YandexCloudUrls {

    private $access_key_id;
    private $secret_access_key;

    public function __construct($key_id, $secret_key) {
      $this->access_key_id = $key_id;
      $this->secret_access_key = $secret_key;
    }

    private function sign($key, $message) {
      return hash_hmac('sha256', $message, $key, true);
    }

    private function getCanonRequest($url, $query) {
      ksort($query);
      return sprintf(
        "GET\n%s\n%s\n%s\n%s\nUNSIGNED-PAYLOAD", 
        $url,
        http_build_query($query),
        "host:storage.yandexcloud.net\n",
        "host",
      );
    }

    private function getStringToSign($req, $time) {
      return sprintf(
        "AWS4-HMAC-SHA256\n%s\n%s\n%s",
        $time->format("Ymd\THis\Z"),
        sprintf("%s/ru-central1/s3/aws4_request", $time->format("Ymd")),
        hash('sha256', $req),
      );
    }

    public function encodeUrl($resource, $time, $expires) {
      $query = array(
        'X-Amz-Algorithm'     => 'AWS4-HMAC-SHA256',
        'X-Amz-Expires'       => $expires,
        'X-Amz-SignedHeaders' => 'host',
        'X-Amz-Date'          => $time->format("Ymd\THis\Z"),
        'X-Amz-Credential'    => sprintf('%s/%s/ru-central1/s3/aws4_request', $this->access_key_id, $time->format("Ymd")),
      );
      
      $canonReq = $this->getCanonRequest($resource, $query);
      $str2sign = $this->getStringToSign($canonReq, $time);
      
      $dateKey    = $this->sign("AWS4" . $this->secret_access_key, $time->format("Ymd"));
      $regionKey  = $this->sign($dateKey, "ru-central1");
      $serviceKey = $this->sign($regionKey, "s3");
      $signingKey = $this->sign($serviceKey, "aws4_request");
      $signature  = bin2hex($this->sign($signingKey, $str2sign));
    
      $query['X-Amz-Signature'] = $signature;
      $result = sprintf("https://storage.yandexcloud.net%s?%s", $resource, http_build_query($query));
    
      // Debugging:
      // echo sprintf("query:\n%s\n", print_r($query, true));
      // echo sprintf("Canonical request:\n%s\n", $canonReq);
      // echo sprintf("String to sign:\n%s\n", $str2sign);
      // echo sprintf(
      //   "dateKey    %s\nregionKey  %s\nserviceKey %s\nsigningKey %s\nsignature  %s\n",
      //   bin2hex($dateKey),
      //   bin2hex($regionKey),
      //   bin2hex($serviceKey),
      //   bin2hex($signingKey),
      //   $signature,
      // );

      return $result;
    }
  }
?>

yc_urls.py

# https://cloud.yandex.ru/docs/storage/concepts/pre-signed-urls

from datetime import datetime, timedelta
import hmac
import hashlib
from urllib import parse
import pprint

access_key_id = 'JK38EXAMPLEAKDID8'
secret_access_key = 'ExamP1eSecReTKeykdokKK38800'

def dict2query(query: dict) -> str:
  return parse.urlencode(sorted([(k, v) for k, v in query.items()], key=lambda x: x[0]))

def sign(key: bytes, message: str) -> bytes:
  return hmac.new(
    key,
    msg = bytes(message, 'latin-1'),
    digestmod = hashlib.sha256
  ).digest()

def getCanonRequest(url: str, query: dict) -> str:

  data = {
    'url': parse.quote(url),
    'query': dict2query(query),
    'canonHeaders': "host:storage.yandexcloud.net\n",
    'signedHeaders': 'host',
  }
  return "GET\n{url}\n{query}\n{canonHeaders}\n{signedHeaders}\nUNSIGNED-PAYLOAD".format_map(data)

def getStringToSign(req: str, time: datetime):
  reqSha256 = hashlib.sha256(bytes(req, 'latin-1')).hexdigest()
  data = {
    'time': "{:%Y%m%dT%H%M%SZ}".format(time),
    'scope': "{:%Y%m%d}/ru-central1/s3/aws4_request".format(time),
    'req': reqSha256
  }
  return "AWS4-HMAC-SHA256\n{time}\n{scope}\n{req}".format_map(data)

def encodeUrl(resource: str, time: datetime, expires: int):
  query = {
    'X-Amz-Algorithm': 'AWS4-HMAC-SHA256',
    'X-Amz-Expires': expires,
    'X-Amz-SignedHeaders': 'host',
    'X-Amz-Date': "{:%Y%m%dT%H%M%SZ}".format(time),
    'X-Amz-Credential': '{}/{:%Y%m%d}/ru-central1/s3/aws4_request'.format(access_key_id, time),
  }
  
  canonReq = getCanonRequest(resource, query)
  str2sign = getStringToSign(canonReq, time)
  
  dateKey    = sign(bytes("AWS4" + secret_access_key, 'latin-1'), "{:%Y%m%d}".format(time))
  regionKey  = sign(dateKey, "ru-central1")
  serviceKey = sign(regionKey, "s3")
  signingKey = sign(serviceKey, "aws4_request")
  signature  = sign(signingKey, str2sign)

  query['X-Amz-Signature'] = signature.hex()
  result = "https://storage.yandexcloud.net{}?{}".format(resource, dict2query(query))

  print('query:\n{}\n'.format(pprint.pformat(query)))
  print('Canonical request:\n{}\n'.format(canonReq))
  print('String to sign:\n{}\n'.format(str2sign))
  print('dateKey    {}\nregionKey  {}\nserviceKey {}\nsigningKey {}\nsignature  {}\n'.format(
    dateKey.hex(),
    regionKey.hex(),
    serviceKey.hex(),
    signingKey.hex(),
    signature.hex(),
  ))
  return result

time = datetime(2019, 8, 1)
path = '/example-bucket/object-for-share.txt'
result = encodeUrl(path, time, 86400)
print('\n{}'.format(result))