signing.py 2.74 KB
Newer Older
1
import datetime
2
import logging
3 4 5 6 7
import pytz

from django import forms
from django.utils import timezone
from django.utils.http import parse_http_date
Agate's avatar
Agate committed
8

9 10 11
import requests
import requests_http_signature

Agate's avatar
Agate committed
12
from . import exceptions, utils
13 14 15

logger = logging.getLogger(__name__)

16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38
#  the request Date should be between now - 30s and now + 30s
DATE_HEADER_VALID_FOR = 30


def verify_date(raw_date):
    if not raw_date:
        raise forms.ValidationError("Missing date header")

    try:
        ts = parse_http_date(raw_date)
    except ValueError as e:
        raise forms.ValidationError(str(e))
    dt = datetime.datetime.utcfromtimestamp(ts)
    dt = dt.replace(tzinfo=pytz.utc)
    delta = datetime.timedelta(seconds=DATE_HEADER_VALID_FOR)
    now = timezone.now()
    if dt < now - delta or dt > now + delta:
        raise forms.ValidationError(
            "Request Date is too far in the future or in the past"
        )

    return dt

39 40

def verify(request, public_key):
41 42
    verify_date(request.headers.get("Date"))

43
    return requests_http_signature.HTTPSignatureAuth.verify(
Agate's avatar
Agate committed
44
        request, key_resolver=lambda **kwargs: public_key, use_auth_header=False
45 46 47 48 49 50 51 52
    )


def verify_django(django_request, public_key):
    """
    Given a django WSGI request, create an underlying requests.PreparedRequest
    instance we can verify
    """
53
    headers = utils.clean_wsgi_headers(django_request.META)
54 55 56 57 58
    for h, v in list(headers.items()):
        # we include lower-cased version of the headers for compatibility
        # with requests_http_signature
        headers[h.lower()] = v
    try:
Agate's avatar
Agate committed
59
        signature = headers["Signature"]
60 61
    except KeyError:
        raise exceptions.MissingSignature
Agate's avatar
Agate committed
62 63
    url = "http://noop{}".format(django_request.path)
    query = django_request.META["QUERY_STRING"]
64
    if query:
Agate's avatar
Agate committed
65
        url += "?{}".format(query)
66
    signature_headers = signature.split('headers="')[1].split('",')[0]
Agate's avatar
Agate committed
67 68
    expected = signature_headers.split(" ")
    logger.debug("Signature expected headers: %s", expected)
69 70 71 72
    for header in expected:
        try:
            headers[header]
        except KeyError:
Agate's avatar
Agate committed
73
            logger.debug("Missing header: %s", header)
74
    request = requests.Request(
Agate's avatar
Agate committed
75 76
        method=django_request.method, url=url, data=django_request.body, headers=headers
    )
77 78 79 80
    for h in request.headers.keys():
        v = request.headers[h]
        if v:
            request.headers[h] = str(v)
Agate's avatar
Agate committed
81
    request.prepare()
82
    return verify(request, public_key)
83 84 85 86 87


def get_auth(private_key, private_key_id):
    return requests_http_signature.HTTPSignatureAuth(
        use_auth_header=False,
Agate's avatar
Agate committed
88 89 90
        headers=["(request-target)", "user-agent", "host", "date", "content-type"],
        algorithm="rsa-sha256",
        key=private_key.encode("utf-8"),
91 92
        key_id=private_key_id,
    )