OBS request signing

obs.py

import binascii
from datetime import datetime
import hashlib
import hmac
import os

import requests

access_key_id = os.getenv('AWS_ACCESS_KEY_ID')
secret_access_key = os.getenv('AWS_SECRET_ACCESS_KEY')


def custom_domain_names_get(bucket):
    response = make_request(bucket, "/?customdomain")
    return response


def custom_domain_name_put(bucket, domain_name, region='eu-de'):
    resource = f"/?customdomain={domain_name}"
    response = make_request(bucket, resource, method="PUT")
    return response


def make_request(bucket, resource, region="eu-de", method="GET", content_type="application/json", canonicalized_headers=""):
    """See https://docs.otc.t-systems.com/api/obs/obs_04_0010.html"""
    endpoint = _endpoint(bucket=bucket, region=region)
    url = f'https://{endpoint}{resource}'
    canonicalized_resource = f'/{bucket}{resource}'
    date = datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')
    string_to_sign = method + "\n\n" + content_type + "\n" + date + "\n" + canonicalized_headers + canonicalized_resource
    signature = _sign(string_to_sign)
    headers = {
        "Authorization": f"OBS {access_key_id}:{signature}",
        "Content-Type": content_type,
        "Date": date,
    }
    return(requests.request(url=url, method=method, headers=headers, verify=False))


def _endpoint(bucket, region):
    return f'{bucket}.obs.{region}.otc.t-systems.com'


def _sign(string_to_sign):
    hashed = hmac.new(secret_access_key.encode('UTF-8'), string_to_sign.encode('UTF-8'), hashlib.sha1)
    base64_hash = binascii.b2a_base64(hashed.digest())[:-1].decode('UTF-8')
    return base64_hash


if __name__ == '__main__':
    # response = custom_domain_name_put('nikos-obs-test.ds4g.net', 'nikos-obs-test.ds4g.net')
    response = custom_domain_names_get('nikos-obs-test.ds4g.net')
    print(response.status_code)
    print(response.headers)
    print(response.text)