import logging
import math
import pprint
import six
from six.moves.urllib.parse
import urlparse
from .exc
import (AlreadyProcessed,
MacMismatch,
MisComputedContentHash,
TokenExpired,
MissingContent)
from .util
import (calculate_mac,
calculate_payload_hash,
calculate_ts_mac,
prepare_header_val,
random_string,
strings_match,
utc_now)
default_ts_skew_in_seconds = 60
log = logging.getLogger(__name__)
class HawkEmptyValue(object):
def __eq__(self, other):
return isinstance(other, self.__class__)
def __ne__(self, other):
return (
not self.__eq__(other))
def __nonzero__(self):
return False
def __bool__(self):
return False
def __repr__(self):
return 'EmptyValue'
EmptyValue = HawkEmptyValue()
class HawkAuthority:
def _authorize(self, mac_type, parsed_header, resource,
their_timestamp=
None,
timestamp_skew_in_seconds=default_ts_skew_in_seconds,
localtime_offset_in_seconds=0,
accept_untrusted_content=
False):
now = utc_now(offset_in_seconds=localtime_offset_in_seconds)
their_hash = parsed_header.get(
'hash',
'')
their_mac = parsed_header.get(
'mac',
'')
mac = calculate_mac(mac_type, resource, their_hash)
if not strings_match(mac, their_mac):
raise MacMismatch(
'MACs do not match; ours: {ours}; '
'theirs: {theirs}'
.format(ours=mac, theirs=their_mac))
check_hash =
True
if 'hash' not in parsed_header:
# The request did not hash its content.
if not resource.content
and not resource.content_type:
# It is acceptable to not receive a hash if there is no content
# to hash.
log.debug(
'NOT calculating/verifying payload hash '
'(no hash in header, request body is empty)')
check_hash =
False
elif accept_untrusted_content:
# Allow the request, even if it has content. Missing content or
# content_type values will be coerced to the empty string for
# hashing purposes.
log.debug(
'NOT calculating/verifying payload hash '
'(no hash in header, accept_untrusted_content=True)')
check_hash =
False
if check_hash:
if not their_hash:
log.info(
'request unexpectedly did not hash its content')
content_hash = resource.gen_content_hash()
if not strings_match(content_hash, their_hash):
# The hash declared in the header is incorrect.
# Content could have been tampered with.
log.debug(
'mismatched content: {content}'
.format(content=repr(resource.content)))
log.debug(
'mismatched content-type: {typ}'
.format(typ=repr(resource.content_type)))
raise MisComputedContentHash(
'Our hash {ours} ({algo}) did not '
'match theirs {theirs}'
.format(ours=content_hash,
theirs=their_hash,
algo=resource.credentials[
'algorithm']))
if resource.seen_nonce:
if resource.seen_nonce(resource.credentials[
'id'],
parsed_header[
'nonce'],
parsed_header[
'ts']):
raise AlreadyProcessed(
'Nonce {nonce} with timestamp {ts} '
'has already been processed for {id}'
.format(nonce=parsed_header[
'nonce'],
ts=parsed_header[
'ts'],
id=resource.credentials[
'id']))
else:
log.warning(
'seen_nonce was None; not checking nonce. '
'You may be vulnerable to replay attacks')
their_ts = int(their_timestamp
or parsed_header[
'ts'])
if math.fabs(their_ts - now) > timestamp_skew_in_seconds:
message = (
'token with UTC timestamp {ts} has expired; '
'it was compared to {now}'
.format(ts=their_ts, now=now))
tsm = calculate_ts_mac(now, resource.credentials)
if isinstance(tsm, six.binary_type):
tsm = tsm.decode(
'ascii')
www_authenticate = (
'Hawk ts="{ts}", tsm="{tsm}", error="{error}"'
.format(ts=now, tsm=tsm, error=message))
raise TokenExpired(message,
localtime_in_seconds=now,
www_authenticate=www_authenticate)
log.debug(
'authorized OK')
def _make_header(self, resource, mac, additional_keys=
None):
keys = additional_keys
if not keys:
# These are the default header keys that you'd send with a
# request header. Response headers are odd because they
# exclude a bunch of keys.
keys = (
'id',
'ts',
'nonce',
'ext',
'app',
'dlg')
header = u
'Hawk mac="{mac}"'.format(mac=prepare_header_val(mac))
if resource.content_hash:
header = u
'{header}, hash="{hash}"'.format(
header=header,
hash=prepare_header_val(resource.content_hash))
if 'id' in keys:
header = u
'{header}, id="{id}"'.format(
header=header,
id=prepare_header_val(resource.credentials[
'id']))
if 'ts' in keys:
header = u
'{header}, ts="{ts}"'.format(
header=header, ts=prepare_header_val(resource.timestamp))
if 'nonce' in keys:
header = u
'{header}, nonce="{nonce}"'.format(
header=header, nonce=prepare_header_val(resource.nonce))
# These are optional so we need to check if they have values first.
if 'ext' in keys
and resource.ext:
header = u
'{header}, ext="{ext}"'.format(
header=header, ext=prepare_header_val(resource.ext))
if 'app' in keys
and resource.app:
header = u
'{header}, app="{app}"'.format(
header=header, app=prepare_header_val(resource.app))
if 'dlg' in keys
and resource.dlg:
header = u
'{header}, dlg="{dlg}"'.format(
header=header, dlg=prepare_header_val(resource.dlg))
log.debug(
'Hawk header for URL={url} method={method}: {header}'
.format(url=resource.url, method=resource.method,
header=header))
return header
class Resource:
"""
Normalized request / response resource.
:param credentials:
A dict of credentials; it must have the keys:
``id``, ``key``,
and ``algorithm``.
See :ref:`sending-request`
for an example.
:type credentials_map: dict
:param url: Absolute URL of the request / response.
:type url: str
:param method: Method of the request / response. E.G. POST, GET
:type method: str
:param content=EmptyValue: Byte string of request / response body.
:type content=EmptyValue: str
:param content_type=EmptyValue: content-type header value
for request / response.
:type content_type=EmptyValue: str
:param always_hash_content=
True:
When
True, ``content``
and ``content_type`` must be provided.
Read :ref:`skipping-content-checks` to learn more.
:type always_hash_content=
True: bool
:param ext=
None:
An external `Hawk`_ string.
If not None, this value will be
signed so that the sender can trust it.
:type ext=
None: str
:param app=
None:
A `Hawk`_ string identifying an external application.
:type app=
None: str
:param dlg=
None:
A `Hawk`_ string identifying a
"delegated by" value.
:type dlg=
None: str
:param timestamp=utc_now():
A unix timestamp integer,
in UTC
:type timestamp: int
:param nonce=
None:
A string that when coupled
with the timestamp will
uniquely identify this request / response.
:type nonce=
None: str
:param seen_nonce=
None:
A callable that returns
True if a nonce has been seen.
See :ref:`nonce`
for details.
:type seen_nonce=
None: callable
.. _`Hawk`:
https://github.com/hueniverse/hawk
"""
def __init__(self, **kw):
self.credentials = kw.pop(
'credentials')
self.credentials[
'id'] = prepare_header_val(self.credentials[
'id'])
self.method = kw.pop(
'method').upper()
self.content = kw.pop(
'content', EmptyValue)
self.content_type = kw.pop(
'content_type', EmptyValue)
self.always_hash_content = kw.pop(
'always_hash_content',
True)
self.ext = kw.pop(
'ext',
None)
self.app = kw.pop(
'app',
None)
self.dlg = kw.pop(
'dlg',
None)
self.timestamp = str(kw.pop(
'timestamp',
None)
or utc_now())
self.nonce = kw.pop(
'nonce',
None)
if self.nonce
is None:
self.nonce = random_string(6)
# This is a lookup function for checking nonces.
self.seen_nonce = kw.pop(
'seen_nonce',
None)
self.url = kw.pop(
'url')
if not self.url:
raise ValueError(
'url was empty')
url_parts = self.parse_url(self.url)
log.debug(
'parsed URL parts: \n{parts}'
.format(parts=pprint.pformat(url_parts)))
self.name = url_parts[
'resource']
or ''
self.host = url_parts[
'hostname']
or ''
self.port = str(url_parts[
'port'])
if kw.keys():
raise TypeError(
'Unknown keyword argument(s): {0}'
.format(kw.keys()))
@property
def content_hash(self):
if not hasattr(self,
'_content_hash'):
raise AttributeError(
'Cannot access content_hash because it has not been generated')
return self._content_hash
def gen_content_hash(self):
if self.content == EmptyValue
or self.content_type == EmptyValue:
if self.always_hash_content:
# Be really strict about allowing developers to skip content
# hashing. If they get this far they may be unintentiionally
# skipping it.
raise MissingContent(
'payload content and/or content_type cannot be '
'empty when always_hash_content is True')
log.debug(
'NOT hashing content')
self._content_hash =
None
else:
self._content_hash = calculate_payload_hash(
self.content, self.credentials[
'algorithm'],
self.content_type)
return self.content_hash
def parse_url(self, url):
url_parts = urlparse(url)
url_dict = {
'scheme': url_parts.scheme,
'hostname': url_parts.hostname,
'port': url_parts.port,
'path': url_parts.path,
'resource': url_parts.path,
'query': url_parts.query,
}
if len(url_dict[
'query']) > 0:
url_dict[
'resource'] =
'%s?%s' % (url_dict[
'resource'],
url_dict[
'query'])
if url_parts.port
is None:
if url_parts.scheme ==
'http':
url_dict[
'port'] = 80
elif url_parts.scheme ==
'https':
url_dict[
'port'] = 443
return url_dict
