Feature/aes signature

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'
+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,7 @@ 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,16 @@
+# 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('AFTERSHIP_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,13 @@
+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 API_KEY_FILED_NAME, API_ENDPOINT, API_VERSION, AS_SIGNATURE_HMAC_SHA256, AS_API_KEY, CONTENT_TYPE, SIGNATURE_AES_HMAC_SHA256 
+from .util import get_api_key, get_api_secret
 
 
 def build_request_url(path):
@@ -12,9 +16,65 @@ 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()
+    elif headers.get(AS_API_KEY) is None:
+        headers[AS_API_KEY] = get_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_api_key()
+
+    body = kwargs.get('json', None)
+    date, sign_string = gen_sign_string(method, path, body, headers, content_type)
+
+    hmac = Hmac(get_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/response.py

@@ -17,7 +17,7 @@ def process_response(response):
         json_content = response.json()
     except json.JSONDecodeError:
         raise InternalError
-
+        
     if response.status_code // 100 == 2:
         return json_content['data']
 

aftership/signstring/signstring.py

@@ -0,0 +1,71 @@
+# coding=utf-8
+
+import os
+import hashlib
+import urllib
+import time
+import json
+from urllib import parse
+from typing import Union,Text,Dict
+# from aftership.const import API_KEY_FILED_NAME, API_ENDPOINT, AS_SIGNATURE_HMAC_SHA256, AS_SIGNATURE_RSA_SHA256, AS_API_KEY 
+
+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('AFTERSHIP_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())
+        # date = "Tue, 29 Jun 2021 07:55:55 GMT"
+        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

@@ -23,3 +23,9 @@ def get_api_key():
     if aftership.api_key is not None:
         return aftership.api_key
     return os.getenv('AFTERSHIP_API_KEY')
+
+def get_api_secret():
+    """Get AfterShip API secret"""
+    if aftership.api_secret is not None:
+        return aftership.api_secret
+    return os.getenv('AFTERSHIP_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)

pyproject.toml

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

tests/test_couriers.py

@@ -3,6 +3,9 @@
 import pytest
 
 from aftership import courier
+from aftership import const
+
+body='{"source_id":"1710452100BO","name":"1710452100BO","number":"1710452100BO","currency":"EUR","status":"closed","financial_status":"paid","fulfillment_status":"fulfilled","order_total":"399","shipping_total":"0","tax_total":"63.71","discount_total":"0","subtotal":"335.29","items":[{"source_id":"230278-01","sku":"230278-01","quantity":1,"unit_weight":{"unit":"kg","value":8.06},"unit_price":"399","discount":"0","tax":"63.71","return_quantity":1,"fulfillable_quantity":1,"source_product_id":"230278-01","source_variant_id":"230278-01","title":"Cinetic Big Ball Multi Floor 2","hs_code":null,"origin_country_region":"DEU","image_urls":["https://dyson-h.assetsadobe2.com/is/image/content/dam/dyson/images/products/primary/230278-01.png"]}],"source_created_at":"2022-08-08T09:34:03.000Z","source_updated_at":"2022-08-08T09:34:03.000Z","customer":{"source_id":"1710452100BO","first_name":"MICH DE EU RETURNS","last_name":"MICH DE EU RETURNS","emails":["[email protected]"],"phones":["3745345234"]},"shipping_address":{"first_name":"Boon","address_line_1":"Albrechtstrasse","city":"Oberhausen","state":"Nordrhein-Westfalen","country_region":"DEU","postal_code":"46145","phone":"3745345234"}}'
 
 
 class CourierTestCase(TestCase):
@@ -12,14 +15,34 @@ def test_get_couriers(self):
         self.assertIn('total', resp)
         self.assertIn('couriers', resp)
 
+    @pytest.mark.vcr()
+    def test_aes_get_couriers(self):
+        resp = courier.list_couriers(signature_type=const.SIGNATURE_AES_HMAC_SHA256)
+        self.assertIn('total', resp)
+        self.assertIn('couriers', resp)
+
+
     @pytest.mark.vcr()
     def test_get_all_couriers(self):
         resp = courier.list_all_couriers()
         self.assertIn('total', resp)
         self.assertIn('couriers', resp)
 
+    @pytest.mark.vcr()
+    def test_aes_get_all_couriers(self):
+        resp = courier.list_all_couriers(signature_type=const.SIGNATURE_AES_HMAC_SHA256)
+        self.assertIn('total', resp)
+        self.assertIn('couriers', resp)
+
     @pytest.mark.vcr()
     def test_detect_courier(self):
-        resp = courier.detect_courier(tracking={'tracking_number': '1234567890'})
+        resp = courier.detect_courier(tracking={"tracking_number": "1234567890"})
         self.assertIn('total', resp)
         self.assertIn('couriers', resp)
+
+    @pytest.mark.vcr()
+    def test_aes_detect_courier(self):
+        resp = courier.detect_courier(signature_type=const.SIGNATURE_AES_HMAC_SHA256,
+                                        tracking={"tracking_number": "1234567890"})
+        self.assertIn('total', resp)
+        self.assertIn('couriers', resp)

tests/test_estimated_delivery_date.py

@@ -3,13 +3,14 @@
 import pytest
 
 from aftership import estimated_delivery_date
+from aftership import const
+
 
 
 class EstimatedDeliveryDateTestCase(TestCase):
     @pytest.mark.vcr()
     def test_batch_predict_estimated_delivery_date(self):
-        resp = estimated_delivery_date.batch_predict_estimated_delivery_date(
-            estimated_delivery_dates=[
+        estimated_delivery_dates=[
                 {
                     "slug": "fedex",
                     "service_type_name": "FEDEX HOME DELIVERY",
@@ -36,5 +37,10 @@ def test_batch_predict_estimated_delivery_date(self):
                         "pickup_time": ""
                     }
                 }
-            ])
+            ]
+        resp = estimated_delivery_date.batch_predict_estimated_delivery_date(estimated_delivery_dates=estimated_delivery_dates)
+        print(resp)
+        resp = estimated_delivery_date.batch_predict_estimated_delivery_date(estimated_delivery_dates=estimated_delivery_dates, 
+                                                                            signature_type=const.SIGNATURE_AES_HMAC_SHA256)
         print(resp)
+

tests/test_notifications.py

@@ -4,18 +4,34 @@
 
 import aftership
 
+from aftership import const
 
 class NotificationTestCase(TestCase):
     @pytest.mark.vcr()
     def test_list_notification(self):
         response = aftership.notification.list_notifications(tracking_id='k5lh7dy7vvqeck71p5loe011')
 
+    @pytest.mark.vcr()
+    def test_aes_list_notification(self):
+        response = aftership.notification.list_notifications(tracking_id='k5lh7dy7vvqeck71p5loe011', 
+        signature_type=const.SIGNATURE_AES_HMAC_SHA256)
+
     @pytest.mark.vcr()
     def test_add_notification(self):
         response = aftership.notification.add_notification(tracking_id='k5lh7dy7vvqeck71p5loe011',
                                                            notification={'emails': ['[email protected]']})
+
+    def test_aes_add_notification(self):
+        response = aftership.notification.add_notification(tracking_id='k5lh7dy7vvqeck71p5loe011',
+                                                           notification={"emails": ["[email protected]"]}, 
+                                                           signature_type=const.SIGNATURE_AES_HMAC_SHA256)
 
     @pytest.mark.vcr()
     def test_remove_notification(self):
         response = aftership.notification.remove_notification(tracking_id='k5lh7dy7vvqeck71p5loe011',
-                                                              notification={'emails': ['[email protected]']})
+                                                              notification={"emails": ["[email protected]"]})
+
+    def test_aes_remove_notification(self):
+        response = aftership.notification.remove_notification(tracking_id='k5lh7dy7vvqeck71p5loe011',
+                                                              notification={"emails": ["[email protected]"]},
+                                                              signature_type=const.SIGNATURE_AES_HMAC_SHA256)