Initial commit

Author: skylander86

Pipfile

@@ -0,0 +1,13 @@
+[[source]]
+name = "pypi"
+url = "https://pypi.org/simple"
+verify_ssl = true
+
+[dev-packages]
+twine = "*"
+
+[packages]
+pynacl = "*"
+
+[requires]
+python_version = "3.8"

Pipfile.lock

@@ -1,2 +1,62 @@
-# formsg-python-sdk
-Python SDK for integrating with FormSG
+# FormSG Python SDK
+
+[![PyPI version](https://img.shields.io/pypi/v/formsg.svg)](https://pypi.python.org/pypi/formsg/)
+[![PyPI license](https://img.shields.io/pypi/l/formsg.svg)](https://pypi.python.org/pypi/formsg/)
+[![PyPI pyversions](https://img.shields.io/pypi/pyversions/formsg.svg)](https://pypi.python.org/pypi/formsg/)
+[![PyPI status](https://img.shields.io/pypi/status/formsg.svg)](https://pypi.python.org/pypi/formsg/)
+[![PyPI download total](https://img.shields.io/pypi/dm/formsg.svg)](https://pypi.python.org/pypi/formsg/)
+
+This SDK provides convenient utilities for verifying FormSG webhooks and decrypting submissions in Python and Flask or Django.
+
+## Installation
+
+```bash
+pip install formsg
+```
+
+## Usage
+
+The SDK provides two main utility functions for handling FormSG webhook:
+
+- [`verify_signature(webhook_uri, signature_header, signature_expiry_seconds=60)`](formsg/webhook.py) verifies that the incoming webhook's signature is valid based on the FormSG production public key.
+It raises a `nacl.exceptions.BadSignatureError` if the signature is invalid.
+The signature header is usually found in the `X-FormSG-Signature` header.
+Details on how the signature is constructed can be found [here](https://github.com/opengovsg/formsg-javascript-sdk/#verifying-signatures-manually).
+
+- [`decrypt_content(body_json, secret_key)`](formsg/webhook.py) will decrypt the encrypted content using the given Base-64 encoded secret key.
+`body_json` is expected to be a dictionary-like object.
+
+For convenience, the SDK implements a [`decrypt_django_request`](formsg/django.py) and [`decrypt_flask_request`](formsg/flask.py) which returns the decrypted FormSG content from a Django/Flask request object directly.
+
+### Example with Flask
+
+```python
+from formsg.webhook import decrypt_flask_request
+
+from flask import Flask
+from flask import jsonify
+from flask import request
+
+app = Flask(__name__)
+
+
+@app.route('/formsg_webhook', methods=['POST'])
+def formsg_webhook():
+    decrypted = decrypt_flask_request(
+        request,
+        secret_key='o4qGJ/AFpmToTOpqptMyTsV3WofQjD7dX6cpVZ7RwNA=',
+        webhook_uri='https://90da680eb8fa.ngrok.io/formsg_webhook',  # we use ngrok to test our webhooks locally
+    )
+
+    return jsonify(decrypted)
+#end def
+
+
+if __name__ == '__main__':
+    app.run(debug=True)
+#end if
+```
+
+## Contributions
+
+If you find any issues or would like to contribute improvements, please feel free to raise them in this repository directly.

README.md

@@ -0,0 +1,3 @@
+from .webhook import *
+from .django import *
+from .flask import *

formsg/__init__.py

@@ -0,0 +1,16 @@
+__all__ = ['decrypt_django_request']
+from .webhook import decrypt_content
+from .webhook import verify_signature
+import json
+
+
+def decrypt_django_request(request, secret_key, webhook_uri=None, signature_expiry_seconds=60):
+    if webhook_uri is None:
+        webhook_uri = request.build_absolute_uri()
+
+    verify_signature(webhook_uri, request.headers['X-FormSG-Signature'], signature_expiry_seconds=signature_expiry_seconds)
+
+    body_json = json.loads(request.body)
+
+    return decrypt_content(body_json, secret_key)
+#end def

formsg/django.py

@@ -0,0 +1,15 @@
+__all__ = ['decrypt_flask_request']
+from .webhook import decrypt_content
+from .webhook import verify_signature
+
+
+def decrypt_flask_request(request, secret_key, webhook_uri=None, signature_expiry_seconds=60):
+    if webhook_uri is None:
+        webhook_uri = request.url
+
+    verify_signature(webhook_uri, request.headers['X-FormSG-Signature'], signature_expiry_seconds=signature_expiry_seconds)
+
+    body_json = request.get_json()
+
+    return decrypt_content(body_json, secret_key)
+#end def

formsg/flask.py

@@ -0,0 +1,63 @@
+__all__ = ['verify_signature', 'decrypt_content']
+import json
+import logging
+import re
+from time import time
+from urllib.parse import urlparse
+
+from nacl.encoding import Base64Encoder
+from nacl.exceptions import BadSignatureError
+from nacl.public import Box
+from nacl.public import PrivateKey
+from nacl.public import PublicKey
+from nacl.signing import VerifyKey
+
+logger = logging.getLogger(__name__)
+
+FORMSG_WEBHOOK_PUBLIC_KEY = VerifyKey('3Tt8VduXsjjd4IrpdCd7BAkdZl/vUCstu9UvTX84FWw=', encoder=Base64Encoder)
+ENCRYPTED_CONTENT_REGEX = re.compile(r'^(?P<submission_public_key>[\w\+\/\=]*)\;(?P<nonce>[\w\+\/\=]*)\:(?P<encrypted_message>[\w\+\/\=]*)$')
+
+
+def verify_signature(webhook_uri, signature_header, signature_expiry_seconds=60):
+    # v1 is signature, s is submissionId, f is formId, t is submission epoch
+    logger.debug(f'X-FormSG-Signature is <{signature_header}>.')
+    formsg_signature = dict(part.split('=', 1) for part in signature_header.split(','))
+    formsg_signature['t'] = int(formsg_signature['t'])
+
+    # Javascript url.href adds a trailing `/` to root domain urls
+    # https://github.com/opengovsg/formsg-javascript-sdk/blob/master/src/webhooks.ts#L25
+    u = urlparse(webhook_uri)
+    if not u.path:
+        u = u._replace(path='/')
+    webhook_uri = u.geturl()
+
+    FORMSG_WEBHOOK_PUBLIC_KEY.verify(
+        smessage=f'{webhook_uri}.{formsg_signature["s"]}.{formsg_signature["f"]}.{formsg_signature["t"]}'.encode('ascii'),
+        signature=Base64Encoder.decode(formsg_signature['v1']),
+    )
+
+    if time() - (formsg_signature['t'] / 1000) > signature_expiry_seconds:
+        raise BadSignatureError('FormSG signature has expired.')
+
+    return formsg_signature
+#end def
+
+
+def decrypt_content(body_json, secret_key):
+    if 'data' in body_json:
+        encrypted_content = body_json['data']['encryptedContent']
+    else:
+        encrypted_content = body_json['encryptedContent']  # old version POST body
+    #end if
+
+    submission_public_key, nonce, encrypted_message = ENCRYPTED_CONTENT_REGEX.match(encrypted_content).groups()
+
+    box = Box(
+        PrivateKey(secret_key, encoder=Base64Encoder),
+        PublicKey(submission_public_key, encoder=Base64Encoder),
+    )
+
+    plaintext = box.decrypt(encrypted_message, Base64Encoder.decode(nonce), encoder=Base64Encoder)
+
+    return json.loads(plaintext)
+#end def