Merge pull request #29 from 420neshot/larky_aes_ctr

New Larky AES_CTR sample

Author: mottersheadt

integrations/larky/crypto/AES_CTR/README.md

@@ -0,0 +1,45 @@
+# AES CTR sample of encryption in Larky
+
+This sample includes:
+1. Python script that generates an ectrypted message of an input data;
+2. Larky test `.star` file that generates the same result of the same input data;
+3. Ready-for-use YAML (Inbound Route) that produces the same result as previous scripts.
+
+## Testing part:
+
+#### 1. Python:
+
+Python script includes all hard-coded inside. As a result, it prints the encrypted message:
+```
+$ python python_sample.py
+JmVuY3J5cHRlZERhdGE9UnMxSHg0NGZROWVFUXpKaVRGSHBSWU93a0NIS0lpZmxLVFhlLXAtWmlBMDhDa3VHWF96c2x3TzROa2Nfd0dZVzlBcVVTenEzT1NyN2RLOEo1c05WU20zQ0xtckU0ejFhNnRCYnE2TjN5dmdnVXhLb1JTVWp4Nnp4Z1dYTm1BZDdrZjVyYlhydlhnR2UwZVZjdlBFQUE3ZkVmbUJYZDdMM2pWT1JKekJIMGdMWDE0Ry1vQ2pFQVg0VmcxQ2xUbnQwZmVva255VXphczVUcHJld1dGOUJoQ1kzN3VYTVM2U2pNalJiRmc9PSZpdj1zZUJoZU5OWXpFVVJ1eVFS
+```
+
+#### 2. Larky test:
+
+To be able to run Larky locally, you'll need to setup your local environment:
+https://www.verygoodsecurity.com/docs/larky/test-larky-locally
+
+Example of run:
+
+<img width="1367" alt="image" src="https://user-images.githubusercontent.com/78090218/189316028-377edbbd-14c2-42be-ad83-5be87a11607d.png">
+
+#### 3. YAML file:
+
+Upload the YAML to your vault and run:
+```
+curl https://tntbmt67sc7.sandbox.verygoodproxy.com/post -k \
+  -H "Content-type: application/json" \
+  -d '{
+    "query": "MID=4111111111111111&exp_year=2030&exp_month=09&cvv=123&name=CoreyTaylor&amount=100",
+    "python": "JmVuY3J5cHRlZERhdGE9UnMxSHg0NGZROWVFUXpKaVRGSHBSWU93a0NIS0lpZmxLVFhlLXAtWmlBMDhDa3VHWF96c2x3TzROa2Nfd0dZVzlBcVVTenEzT1NyN2RLOEo1c05WU20zQ0xtckU0ejFhNnRCYnE2TjN5dmdnVXhLb1JTVWp4Nnp4Z1dYTm1BZDdrZjVyYlhydlhnR2UwZVZjdlBFQUE3ZkVmbUJYZDdMM2pWT1JKekJIMGdMWDE0Ry1vQ2pFQVg0VmcxQ2xUbnQwZmVva255VXphczVUcHJld1dGOUJoQ1kzN3VYTVM2U2pNalJiRmc9PSZpdj1zZUJoZU5OWXpFVVJ1eVFS"
+    }'
+```
+
+Example of response:
+```
+"json": {
+    "larky_": "JmVuY3J5cHRlZERhdGE9UnMxSHg0NGZROWVFUXpKaVRGSHBSWU93a0NIS0lpZmxLVFhlLXAtWmlBMDhDa3VHWF96c2x3TzROa2Nfd0dZVzlBcVVTenEzT1NyN2RLOEo1c05WU20zQ0xtckU0ejFhNnRCYnE2TjN5dmdnVXhLb1JTVWp4Nnp4Z1dYTm1BZDdrZjVyYlhydlhnR2UwZVZjdlBFQUE3ZkVmbUJYZDdMM2pWT1JKekJIMGdMWDE0Ry1vQ2pFQVg0VmcxQ2xUbnQwZmVva255VXphczVUcHJld1dGOUJoQ1kzN3VYTVM2U2pNalJiRmc9PSZpdj1zZUJoZU5OWXpFVVJ1eVFS",
+    "python": "JmVuY3J5cHRlZERhdGE9UnMxSHg0NGZROWVFUXpKaVRGSHBSWU93a0NIS0lpZmxLVFhlLXAtWmlBMDhDa3VHWF96c2x3TzROa2Nfd0dZVzlBcVVTenEzT1NyN2RLOEo1c05WU20zQ0xtckU0ejFhNnRCYnE2TjN5dmdnVXhLb1JTVWp4Nnp4Z1dYTm1BZDdrZjVyYlhydlhnR2UwZVZjdlBFQUE3ZkVmbUJYZDdMM2pWT1JKekJIMGdMWDE0Ry1vQ2pFQVg0VmcxQ2xUbnQwZmVva255VXphczVUcHJld1dGOUJoQ1kzN3VYTVM2U2pNalJiRmc9PSZpdj1zZUJoZU5OWXpFVVJ1eVFS"
+  }
+```

integrations/larky/crypto/AES_CTR/inbound-AES-CTR.yml

@@ -0,0 +1,100 @@
+data:
+  - attributes:
+      created_at: '2021-04-26T09:55:37'
+      destination_override_endpoint: 'https://echo.apps.verygood.systems'
+      entries:
+        - classifiers: {}
+          config:
+            condition: AND
+            rules:
+              - condition: null
+                expression:
+                  field: PathInfo
+                  operator: matches
+                  type: string
+                  values:
+                    - /post
+              - condition: null
+                expression:
+                  field: ContentType
+                  operator: equals
+                  type: string
+                  values:
+                    - application/json
+          id: 3e4ca047-23e2-4397-9a67-d673cedf4cc8
+          id_selector: null
+          operation: REDACT
+          operations:
+            - name: github.com/verygoodsecurity/common/compute/larky/http/Process
+              parameters:
+                script: |-
+                  load("@stdlib//builtins", builtins="builtins")
+                  load("@stdlib//json", json="json")
+                  load("@vgs//vault", "vault")
+                  load("@stdlib//hashlib", "hashlib")
+                  load("@stdlib//binascii", "binascii")
+                  load("@vgs//vault", "vault")
+                  load("@stdlib//base64", base64="base64")
+                  load("@vendor//Crypto/Hash/HMAC", HMAC="HMAC")
+                  load("@stdlib//string", string="string")
+                  load("@stdlib//random", random="random") # instead of 'secrets'
+                  load("@vendor//Crypto/Cipher/AES", AES="AES")
+                  load("@vendor//Crypto/Util/Counter", Counter="Counter") # correct?
+                  load("@vendor//Crypto/Hash/SHA256", SHA256="SHA256")
+
+                  def process(input, ctx):
+                    API_SECRET = '5243A220F218587596BC3A9E3C47A4E7B4FF98F7136856F174940D22071A35DD'
+                    secondKey = 'MDIzM1OCNkMzRDTJGNDIc0MM1MkEz0NTA0E3Rjk5NDc='
+
+                    body_str = str(input.body)
+                    body = json.loads(body_str)
+                    query = body['query']
+
+                    # hashpart
+                    secret = bytes(API_SECRET, encoding='utf-8')
+                    query_tohash = query + API_SECRET
+                    hashValue = HMAC.new(secret, bytes(query_tohash, encoding='utf-8'), digestmod=SHA256)
+                    print (hashValue.hexdigest())
+
+                    # Aes CTR encryption part
+                    to_encrypt = query + '&alg=SHA256&hashed_query='+hashValue.hexdigest()
+                    ivv = 'seBheNNYzEURuyQR' # made it static to compare with Python
+                    secondKey = base64.b64decode(secondKey)
+                    h = binascii.hexlify(bytes(ivv,encoding='utf-8'))
+                    ctr = Counter.new(128, initial_value=int(str(h), 16))
+                    chiper = AES.new(secondKey, AES.MODE_CTR, counter=ctr)
+                    chipertext = chiper.encrypt(bytes(to_encrypt,encoding='utf-8'))
+                    encString = base64.urlsafe_b64encode(chipertext).decode()
+                    payload =  '&encryptedData=' + encString + '&iv=' + ivv
+
+                    # final result to be send to dev.finexusgroup.com
+                    enc = bytes(payload, encoding='utf-8')
+                    final = base64.b64encode(enc).decode()
+
+                    body['larky_'] = final
+                    body.pop('query')
+
+                    input.body = builtins.bytes(json.dumps(body))
+                    return input
+          phase: REQUEST
+          public_token_generator: UUID
+          targets:
+            - body
+          token_manager: PERSISTENT
+          transformer: JSON_PATH
+          transformer_config:
+            - $.email
+          transformer_config_map: null
+      host_endpoint: (.*)\.verygoodproxy\.com
+      id: c891d346-0e1b-49b8-99ac-0754d34f14f2
+      ordinal: null
+      port: 80
+      protocol: http
+      source_endpoint: '*'
+      tags:
+        name: echo.apps.verygood.systems-steel-blue-parallelogram
+        source: RouteContainer
+      updated_at: '2021-05-07T11:45:32'
+    id: c891d346-0e1b-49b8-99ac-0754d34f14f2
+    type: rule_chain
+version: 1

integrations/larky/crypto/AES_CTR/python_sample.py

@@ -0,0 +1,31 @@
+import hmac
+import hashlib
+import base64
+import secrets
+from Crypto.Cipher import AES
+from Crypto.Util import Counter
+import binascii
+from base64 import b64encode
+import string
+API_SECRET = '5243A220F218587596BC3A9E3C47A4E7B4FF98F7136856F174940D22071A35DD'
+query = 'MID=4111111111111111&exp_year=2030&exp_month=09&cvv=123&name=CoreyTaylor&amount=100'
+secondKey = 'MDIzM1OCNkMzRDTJGNDIc0MM1MkEz0NTA0E3Rjk5NDc='
+
+"""hashpart"""
+secret = API_SECRET.encode()
+query_tohash = query + API_SECRET
+hashValue = hmac.new(secret,bytes(query_tohash, encoding='utf-8'),digestmod=hashlib.sha256)
+
+"""Aes CTR encryption part"""
+to_encrypt = query + '&alg=SHA256&hashed_query=' + hashValue.hexdigest()
+secondKey = base64.b64decode(secondKey)
+ivv = 'seBheNNYzEURuyQR' 
+h = binascii.hexlify(bytes(ivv,encoding='utf-8'))
+ctr = Counter.new(128, initial_value=int(h, 16))
+chiper = AES.new(secondKey, AES.MODE_CTR, counter=ctr)
+chipertext = chiper.encrypt(bytes(to_encrypt,encoding='utf-8'))
+encString = base64.urlsafe_b64encode(chipertext).decode()
+payload =  '&encryptedData='+encString+'&iv='+ivv
+
+""" final result to be send"""
+print(base64.b64encode(payload.encode()).decode())

integrations/larky/crypto/AES_CTR/test_aes_ctr.star

@@ -0,0 +1,71 @@
+load("@stdlib//unittest", "unittest")
+load("@vendor//asserts", "asserts")
+load("@vgs//http/request", "VGSHttpRequest")
+load("@stdlib//json", json="json")
+load("@stdlib//builtins", builtins="builtins")
+load("@stdlib//hashlib", "hashlib")
+load("@stdlib//binascii", "binascii")
+load("@vgs//vault", "vault")
+load("@stdlib//base64", base64="base64")
+load("@vendor//Crypto/Hash/HMAC", HMAC="HMAC")
+load("@stdlib//string", string="string")
+load("@stdlib//random", random="random") # instead of 'secrets'
+load("@vendor//Crypto/Cipher/AES", AES="AES")
+load("@vendor//Crypto/Util/Counter", Counter="Counter") # correct?
+load("@vendor//Crypto/Hash/SHA256", SHA256="SHA256")
+
+
+def process(input, ctx):
+    API_SECRET = '5243A220F218587596BC3A9E3C47A4E7B4FF98F7136856F174940D22071A35DD'
+    secondKey = 'MDIzM1OCNkMzRDTJGNDIc0MM1MkEz0NTA0E3Rjk5NDc='
+
+    body_str = str(input.body)
+    body = json.loads(body_str)
+    query = body['query']
+
+    """hashpart"""
+    secret = bytes(API_SECRET, encoding='utf-8')
+    query_tohash = query+API_SECRET
+    hashValue = HMAC.new(secret, bytes(query_tohash, encoding='utf-8'), digestmod=SHA256)
+    print (hashValue.hexdigest())
+
+    """Aes CTR encryption part"""
+    to_encrypt = query + '&alg=SHA256&hashed_query='+hashValue.hexdigest()
+    ivv = 'seBheNNYzEURuyQR' # made it static to compare with Python
+    secondKey = base64.b64decode(secondKey)
+    h = binascii.hexlify(bytes(ivv,encoding='utf-8'))
+    ctr = Counter.new(128, initial_value=int(str(h), 16))
+    chiper = AES.new(secondKey, AES.MODE_CTR, counter=ctr)
+    chipertext = chiper.encrypt(bytes(to_encrypt,encoding='utf-8'))
+    encString = base64.urlsafe_b64encode(chipertext).decode()
+    payload =  '&encryptedData='+encString+'&iv='+ivv
+
+    """ final result to be send"""
+    enc = bytes(payload, encoding='utf-8')
+    final = base64.b64encode(enc).decode()
+
+    body['larky_'] = final
+    body.pop('query')
+
+    input.body = builtins.bytes(json.dumps(body))
+    return input
+
+
+def test_process():
+    headers = {}
+    body = b'{"query": "MID=4111111111111111&exp_year=2030&exp_month=09&cvv=123&name=CoreyTaylor&amount=100"}'
+    input = VGSHttpRequest("https://test.com", data=body, headers=headers, method='POST')
+    response = process(input, None)
+    expected_body = b'{"larky_":"JmVuY3J5cHRlZERhdGE9UnMxSHg0NGZROWVFUXpKaVRGSHBSWU93a0NIS0lpZmxLVFhlLXAtWmlBMDhDa3VHWF96c2x3TzROa2Nfd0dZVzlBcVVTenEzT1NyN2RLOEo1c05WU20zQ0xtckU0ejFhNnRCYnE2TjN5dmdnVXhLb1JTVWp4Nnp4Z1dYTm1BZDdrZjVyYlhydlhnR2UwZVZjdlBFQUE3ZkVmbUJYZDdMM2pWT1JKekJIMGdMWDE0Ry1vQ2pFQVg0VmcxQ2xUbnQwZmVva255VXphczVUcHJld1dGOUJoQ1kzN3VYTVM2U2pNalJiRmc9PSZpdj1zZUJoZU5OWXpFVVJ1eVFS"}'
+    print(response.body)
+    print(expected_body)
+    asserts.assert_that(response.body).is_equal_to(expected_body)
+
+
+def _testsuite():
+  _suite = unittest.TestSuite()
+  _suite.addTest(unittest.FunctionTestCase(test_process))
+  return _suite
+
+_runner = unittest.TextTestRunner()
+_runner.run(_testsuite())