Digital Signature in Python

digital_signature.py

#######################################################################
# 1. Generate public and private keys
#    ```shell
#    openssl genpkey -algorithm ED25519 -out ed25519.pem
#    openssl pkey -in ed25519.pem -pubout > ed25519.pem.pub
#    ```
# 2. Register the public key with the remote service
# 3. Ensure python requirements
#    ```shell
#    pip install requests cryptography
#    ```
# 4. Invoke:
#   ```shell
#   python3 digital_signature.py ed25519.pem 'myKeyId' 'https://example.com/api/v1/thing?x=a'
#   ```
#######################################################################

import json, requests, hashlib, base64, email, sys
from urllib.parse import urlparse
from requests.auth import AuthBase
from cryptography.hazmat.primitives import serialization

def pretty_print(req):
    """
    At this point it is completely built and ready
    to be fired; it is "prepared".

    However pay attention at the formatting used in
    this function because it is programmed to be pretty
    printed and may differ from the actual request.
    """
    print('-----------START-----------')
    print(req.method + ' ' + req.url)
    print('\n'.join('{}: {}'.format(k, v) for k, v in req.headers.items()))
    if req.body:
        print('body\n: ' + req.body)
    print('------------END------------')


def b64str(bs):
    return base64.b64encode(bs).decode(encoding='utf-8')

def body_digest(request):
    bs = None
    digest = None
    if request.body:
        if isinstance(request.body, str):
            bs = bytes(request.body, encoding='utf-8')
        elif isinstance(request.body, bytes):
            bs = request.body
    if bs:
        digest = b64str(hashlib.sha256(bs).digest())
    return digest

def fingerprint(request):
    # NB request.path_url has the url encoded query string
    url = urlparse(request.url)
    ans = "(request-target): " + request.method.lower() + " " + request.path_url + "\n"
    ans = ans + "host: " + url.netloc  + "\n"
    ans = ans + "date: " + request.headers['date']
    if 'digest' in request.headers:
        ans = ans + "\ndigest: " + request.headers['digest'];
    return ans

def sign(req_fp, private_key):
    bs = bytes(req_fp, encoding='utf-8')
    return b64str(private_key.sign(bs))

def intercept(request, key_id, private_key):
    request.headers['date'] = email.utils.formatdate(timeval=None, localtime=False, usegmt=True)
    digest = body_digest(request)
    headers = ['(request-target)', 'host', 'date']
    if digest:
        request.headers['digest'] = 'sha-256=' + digest
        headers.append('digest')
    req_fp = fingerprint(request)
    request_signature = sign(req_fp, private_key)
    request.headers['signature'] = 'keyId="' + key_id + '",algorithm="ed25519"' + ',headers="' + ' '.join(headers) + '",signature="' + request_signature + '"'
    # pretty_print(request)
    return request


def load_private_key(file):
    data = None
    with open(file, 'r') as f:
        data = bytes(f.read(), encoding='utf-8')
    return serialization.load_pem_private_key(data, None)


class SignatureAuth(AuthBase):
    def __init__(self, key_id, private_key):
        self.key_id = key_id
        self.private_key = private_key

    def __call__(self, request):
        return intercept(request, self.key_id, self.private_key)


private_key_file = sys.argv[1]
private_key = load_private_key(private_key_file)
key_id = sys.argv[2]
endpoint = sys.argv[3]
signature_auth = SignatureAuth(key_id, private_key)

response = requests.get(endpoint, auth=signature_auth)
print(response.json())