Feature/aes signature

README.rst

@@ -73,7 +73,7 @@ The following code gets list of supported couriers
  aftership.api_key = 'YOUR_API_KEY_FROM_AFTERSHIP'
  couriers = aftership.courier.list_couriers()
 
-You can also set API key via setting :code:`AFTERSHIP_API_KEY` environment varaible.
+You can also set API key via setting :code:`AFTERSHIP_API_KEY` environment variable.
 
 .. code-block:: bash
 

aftership/__init__.py

@@ -4,3 +4,5 @@
 __version__ = '1.3.0'
 
 api_key = None
+
+api_secret = None

aftership/const.py

@@ -1,4 +1,10 @@
-API_KEY_FILED_NAME = 'aftership-api-key'
+AFTERSHIP_API_KEY = 'aftership-api-key'
+AS_API_KEY = 'as-api-key'
+AS_SIGNATURE_HMAC_SHA256 = 'as-signature-hmac-sha256'
+
+CONTENT_TYPE = "application/json"
 
 API_VERSION = "v4"
 API_ENDPOINT = "https://api.aftership.com/v4/"
+
+SIGNATURE_AES_HMAC_SHA256 = "AES-HMAC-SHA256"

aftership/courier.py

@@ -1,5 +1,6 @@
 from .request import make_request
 from .response import process_response
+import json
 
 __all__ = ['list_couriers', 'list_all_couriers', 'detect_courier']
 
@@ -24,3 +25,8 @@ def detect_courier(tracking, **kwargs):
  """
  response = make_request('POST', 'couriers/detect', json=dict(tracking=tracking), **kwargs)
  return process_response(response)
+
+
+def post_orders(order, **kwargs):
+ response = make_request('POST', 'commerce/v1/orders', json=dict(order=json.loads(order)), **kwargs)
+ return process_response(response)

aftership/hmac/hmac.py

@@ -0,0 +1,18 @@
+# coding=utf-8
+
+import os
+import hashlib
+import base64
+import hmac
+
+
+class Hmac():
+ def __init__(self, api_secret):
+ self.api_secret = api_secret
+
+ def hmac_signature(self, sign_string: str) -> str:
+ if self.api_secret is None:
+ self.api_secret = os.getenv('AS_API_SECRET')
+ signature_str = hmac.new(bytes(self.api_secret.encode()), msg=bytes(
+ sign_string.encode()), digestmod=hashlib.sha256).digest()
+ return base64.b64encode(signature_str).decode()

aftership/request.py

@@ -1,9 +1,14 @@
+import requests
+
 from urllib.parse import urljoin
+from urllib.parse import urlparse
+from aftership.hmac.hmac import Hmac
 
-import requests
+from aftership.signstring.signstring import SignString
 
-from .const import API_KEY_FILED_NAME, API_ENDPOINT
-from .util import get_api_key
+from .const import AFTERSHIP_API_KEY, API_ENDPOINT, AS_SIGNATURE_HMAC_SHA256
+from .const import AS_API_KEY, CONTENT_TYPE, SIGNATURE_AES_HMAC_SHA256
+from .util import get_aftership_api_key, get_as_api_key, get_as_api_secret
 
 
 def build_request_url(path):
@@ -12,9 +17,67 @@ def build_request_url(path):
 
 def make_request(method, path, **kwargs):
  url = build_request_url(path)
+ res = urlparse(url)
+ params = kwargs.get('params', None)
+ path = res.path
+
+ params_str = ""
+ if params:
+ params_str = '&'.join([str(key)+'='+str(value) for key, value in params.items()])
+
+ if not path.startswith("/"):
+ path = '/' + path
+
+ if len(params_str) > 0:
+ path = '{}?{}'.format(path, params_str)
 
+ signature_type = kwargs.pop('signature_type', None)
+ if signature_type is None:
+ return request_with_token(method, url, **kwargs)
+
+ body = kwargs.get('json', None)
+ content_type = None
+ if (method == "POST" or method == "PUT" or method == "PATCH") and body is not None:
+ content_type = CONTENT_TYPE
+
+ # if using SignString, you must use AS_API_KEY header
+ if signature_type == SIGNATURE_AES_HMAC_SHA256:
+ return request_with_aes_hmac256_signature(method, url, path, content_type, **kwargs)
+
+ return None
+
+
+def request_with_token(method, url, **kwargs):
  headers = kwargs.pop('headers', dict())
- if headers.get(API_KEY_FILED_NAME) is None:
- headers[API_KEY_FILED_NAME] = get_api_key()
+ if headers.get(AFTERSHIP_API_KEY) is None and headers.get(AS_API_KEY) is None:
+ headers[AFTERSHIP_API_KEY] = get_aftership_api_key()
+
  kwargs['headers'] = headers
  return requests.request(method, url, **kwargs)
+
+
+def request_with_aes_hmac256_signature(method, url, path, content_type, **kwargs):
+ headers = kwargs.pop('headers', dict())
+ if headers.get(AS_API_KEY, None) is None:
+ headers[AS_API_KEY] = get_as_api_key()
+ if headers.get(AS_API_KEY, None) is None:
+ return
+
+ body = kwargs.get('json', None)
+ date, sign_string = gen_sign_string(method, path, body, headers, content_type)
+
+ hmac = Hmac(get_as_api_secret())
+ hmac_signature = hmac.hmac_signature(sign_string)
+ headers[AS_SIGNATURE_HMAC_SHA256] = hmac_signature
+ headers['Date'] = date
+
+ if content_type is not None:
+ headers["Content-Type"] = content_type
+
+ kwargs['headers'] = headers
+ return requests.request(method, url, **kwargs)
+
+
+def gen_sign_string(method, path, body, headers, content_type):
+ s = SignString(headers[AS_API_KEY])
+ return s.gen_sign_string(method, path, body, headers, content_type)

aftership/signstring/signstring.py

@@ -0,0 +1,69 @@
+# coding=utf-8
+
+import hashlib
+import urllib
+import time
+import json
+from urllib import parse
+from typing import Union, Text, Dict
+
+BODY = Union[Text, Dict]
+
+
+class SignString():
+ def __init__(self, api_secret: str):
+ self.api_secret = api_secret
+
+ def _gen_sign_string(self, method: str, body: BODY, content_type: str, date: str, canonicalized_as_headers: str,
+ canonicalized_resource: str) -> str:
+ result = ''
+ result += method + '\n'
+ if body:
+ if isinstance(body, dict):
+ body = json.dumps(body)
+ body = hashlib.md5(body.encode()).hexdigest().upper()
+ else:
+ body = ''
+ content_type = ''
+
+ result += body + '\n'
+ result += content_type + '\n'
+ result += date + '\n'
+ result += canonicalized_as_headers + '\n'
+ result += canonicalized_resource
+
+ return result
+
+ def _get_canonicalized_as_headers(self, headers: dict) -> str:
+ new_header = {}
+ for k, v in headers.items():
+ new_key = k.lower()
+ new_value = v.strip()
+ new_header.update({new_key: new_value})
+
+ new_header = dict(sorted(new_header.items()))
+
+ result = '\n'.join([k + ':' + v for k, v in new_header.items()])
+ return result
+
+ def _get_canonicalized_resource(self, raw_url: str) -> str:
+ url_parse_result = parse.urlsplit(raw_url)
+ path = url_parse_result.path
+ query = urllib.parse.urlencode(sorted(dict(parse.parse_qsl(url_parse_result.query)).items()))
+ if query:
+ path = path + '?' + query
+ return path
+
+ def gen_sign_string(self, method: str, uri: str, body: str, as_header: dict, content_type: str) -> tuple:
+ # if self.api_secret is None:
+ # self.api_secret = os.getenv('AS_API_SECRET')
+
+ canonicalized_as_headers = self._get_canonicalized_as_headers(as_header)
+ canonicalized_resource = self._get_canonicalized_resource(uri)
+
+ date = time.strftime('%a, %d %b %Y %H:%M:%S GMT', time.gmtime())
+ sign_string = self._gen_sign_string(method=method, body=body, content_type=content_type, date=date,
+ canonicalized_as_headers=canonicalized_as_headers,
+ canonicalized_resource=canonicalized_resource)
+
+ return date, sign_string

aftership/util.py

@@ -18,8 +18,22 @@ def _build_tracking_url(tracking_id, slug, tracking_number):
  return url
 
 
-def get_api_key():
+def get_aftership_api_key():
  """Get AfterShip API key"""
  if aftership.api_key is not None:
  return aftership.api_key
  return os.getenv('AFTERSHIP_API_KEY')
+
+
+def get_as_api_key():
+ """Get AS API key"""
+ if aftership.api_key is not None:
+ return aftership.api_key
+ return os.getenv('AS_API_KEY')
+
+
+def get_as_api_secret():
+ """Get AfterShip API secret"""
+ if aftership.api_secret is not None:
+ return aftership.api_secret
+ return os.getenv('AS_API_SECRET')

examples/courier_example_aes_signature.py

@@ -0,0 +1,21 @@
+import aftership
+
+aftership.api_key = 'asak_61f86xxx'
+aftership.api_secret = 'assk_e8b4bxxx'
+
+
+def get_enabled_courier_names():
+ result = aftership.courier.list_couriers(signature_type="AES")
+ courier_list = [courier['name'] for courier in result['couriers']]
+ return courier_list
+
+
+def get_supported_courier_names():
+ result = aftership.courier.list_all_couriers(signature_type="AES")
+ courier_list = [courier['name'] for courier in result['couriers']]
+ return courier_list
+
+
+if __name__ == '__main__':
+ enabled_couriers = get_enabled_courier_names(signature_type="AES")
+ print(enabled_couriers)

examples/estimated_delivery_date_example.py

@@ -47,6 +47,7 @@ def batch_predict_estimated_delivery_date():
  )
  return result['estimated_delivery_dates']
 
+
 if __name__ == '__main__':
  list = batch_predict_estimated_delivery_date()
- print(list)
+ print(list)

examples/notification_example.py

@@ -17,10 +17,18 @@ def add_notification(tracking_id, emails=None, smses=None, webhook=None):
  raise ValueError('You must specify one of emails, smses or webhook')
 
  aftership.notification.add_notification(tracking_id=tracking_id, notification=update_params)
+
+ ## using HMAC-SHA256 signature
+ aftership.notification.add_notification(tracking_id=tracking_id, notification=update_params, signature_type=const.SIGNATURE_AES_HMAC_SHA256)
+
+
 
 
 def list_notifications(tracking_id):
  notification = aftership.notification.list_notifications(tracking_id=tracking_id)
+
+ ## using HMAC-SHA256 signature
+ notification = aftership.notification.list_notifications(tracking_id=tracking_id, signature_type=const.SIGNATURE_AES_HMAC_SHA256)
  return notification
 
 

examples/tracking_example.py

@@ -1,4 +1,5 @@
 import aftership
+from aftership import const
 
 
 # aftership.api_key = 'PUT_YOUR_AFTERSHIP_KEY_HERE'
@@ -14,6 +15,11 @@ def create_tracking(slug, tracking_number):
 
 def update_tracking(tracking_id, **values):
  aftership.tracking.update_tracking(tracking_id=tracking_id, tracking=values)
+
+ # using HMAC-SHA256 signature
+ aftership.tracking.update_tracking(tracking_id=tracking_id, tracking=values,
+ signature_type=const.SIGNATURE_AES_HMAC_SHA256)
+
  return True
 
 

pyproject.toml

@@ -32,7 +32,7 @@ packages = [
 
 [tool.poetry.dependencies]
 python = "^3.6"
-requests = "^2.0.0'"
+requests = "^2.0.0"
 
 [tool.poetry.dev-dependencies]
 pytest = "^6.0.1"

tests/conftest.py

@@ -6,7 +6,7 @@ def vcr_config():
  return {
  'cassette_library_dir': 'tests/fixtures/cassettes',
  'serializer': 'yaml',
- 'filter_headers': [('aftership-api-key', 'YOUR_API_KEY_IS_HERE')],
+ 'filter_headers': [('aftership-api-key', 'YOUR_API_KEY_IS_HERE'),('as-api-key', 'YOUR_AS_API_KEY'), ('as-api-ssecret', 'YOUR_AS_API_SECRET')],
  'record_mode': 'none',
  'match_on': ['uri', 'method', 'query', 'body'],
  }