Quick Python script to demo calling the NHS MESH API

meshapi.py

import requests
import uuid
import datetime
import hmac
import hashlib
import logging

#This is the set of headers required for an authenticate request.
MESH_HEADERS = {
     "Authorization":"",
     "Mex-ClientVersion": "1",
     "Mex-OSVersion": "10",
     "Mex-OSName": "Linux"
}

#This is the OpenTest MESH Endpoint. The DNS name resolves to 192.168.128.11
MESH_ENDPOINT = "https://msg.opentest.hscic.gov.uk/messageexchange" 
#This is the ID of the MESH Mailbox issued by NHSd
MESH_MAILBOX = "MAILBOX ID "
#This is the password issued for the mailbox above. Issued by NHSd.
MESH_PASSWORD = "MAILBOX PASSWORD"
#This is the shared secret used to sign the hash in the authentication token. Issued by NHSd.
MESH_HASH_SECRET = "MESH HASH SECRET"
#This is the path to your PEM formated client certificate. A certificate used to access other Spine services will work.
#Otherwise it's a client certificate specifically issued for MESH.
SPINE_OR_MESH_CERT = "PATH TO CLIENT CERTIFICATE"
#This is the path to your PEM formated client key. A key used to access other Spine services will work.
#Otherwise it's a client certificate key issued for MESH.
SPINE_OR_MESH_KEY = "PATH TO CLIENT KEY"
#This is the bundle of certificates for the spine environment. Used to validate the TLS certificate returned from the server.
SPINE_CA_BUNDLE = "CERTIFICATE CHAIN FOR SPINE ENVIRONMENT"

#A basic class to wrap a MESH error.
class MeshException(Exception):
    def __init__(self, message, code):
        super().__init__(message)
        self.code = code

#If the response has status code 200 then return it. Otherwise raise a MESH exception.
def response_or_exception(response):
    if response.status_code==200:
        return response
    raise(MeshException(response.text,response.status_code))

"""
Make the hash signature for the authorisation token. It is an HMAC/SHA256 hash of these fields joined by a colon (:) :
Mailbox ID - ID of the Mailbox sending the HTTP Request, must be uppercase.
Nonce - A GUID used as an encryption 'Nonce'
NonceCount - The number of times that the same 'Nonce' has been used.
Mailbox Password - The password for the mesh mailbox
Timestamp - The current date and time in 'yyyyMMddHHmm' format.
"""
def make_mesh_hash(token):
    signature = hmac.new(
        MESH_HASH_SECRET.encode("UTF-8"),
        msg=token.encode("UTF-8"),
        digestmod=hashlib.sha256
    ).hexdigest()
    logging.debug("Hashed Token:%s" % signature)
    return signature

"""
Make the mesh authentication token to go in the authorisation header. 
It is a string of these fields joined by a colon (:) and prefixed with NHSMESH :
Mailbox ID - ID of the Mailbox sending the HTTP Request, must be uppercase.
Nonce - A GUID used as an encryption 'Nonce'
NonceCount - The number of times that the same 'Nonce' has been used.
Timestamp - The current date and time in 'yyyyMMddHHmm' format.
Token Signature - As generated above.
"""
def make_mesh_token():
    nonce = str(uuid.uuid4())
    nonce_count = "001"
    timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M")
    token_body = "%s:%s:%s:%s" % (MESH_MAILBOX,nonce,nonce_count,timestamp)
    plain_hash = "%s:%s:%s:%s:%s" % (MESH_MAILBOX,nonce,nonce_count,MESH_PASSWORD,timestamp)
    mesh_hash = make_mesh_hash(plain_hash)
    token = "NHSMESH %s:%s" % (token_body,mesh_hash)
    logging.debug("MESH Token:%s" % token)
    return token

"""
Make a request to MESH. Include the authentication header. Set up the path and use the correct http verb.
Return a requests Response object if the status code is 200.
Raise a MESHException if the status code is not 200.
"""
def make_mesh_request(url_part,headers={}, do_post=False, do_put=False):
    if (len(url_part)>0 and not url_part.startswith("/")):
        url_part = "/%s" % (url_part,)
    url = "%s/%s%s" % (MESH_ENDPOINT,MESH_MAILBOX,url_part)
    logging.debug("Making call to %s",url)
    token = make_mesh_token()
    headers["Authorization"] = token
    if(do_post):
        response = requests.post(url,headers=headers,verify=SPINE_CA_BUNDLE,
                cert=(SPINE_OR_MESH_CERT, SPINE_OR_MESH_KEY))
    elif(do_put):
        response = requests.put(url,headers=headers,verify=SPINE_CA_BUNDLE,
                cert=(SPINE_OR_MESH_CERT, SPINE_OR_MESH_KEY))
    else:
        response = requests.get(url,headers=headers,verify=SPINE_CA_BUNDLE,
                    cert=(SPINE_OR_MESH_CERT, SPINE_OR_MESH_KEY))
    return response_or_exception(response)

"""
POST an authenticate request to the root url using the full set of MESH headers required.
This should be the first call made to ensure connectivity.
"""
def authenticate():
    response = make_mesh_request("",MESH_HEADERS,do_post=True)
    logging.info("%s %s",response.status_code,response.text)

"""
Make a GET request to retrieve a list of message in the inbox to the /inbox url. 
No extra headers required.
"""
def check_inbox():
    response = make_mesh_request("inbox")
    logging.info("%s %s",response.status_code,response.text)

"""
Make a GET request to retrieve a count of messages in the inbox to the /count url. 
No extra headers required.
"""
def check_inbox_count():
    response = make_mesh_request("count")
    logging.info("%s %s",response.status_code,response.text)

"""
Make a GET request to retrieve a the first (or only) chunk of a message from the inbox.
Uses the url /inbox/{messageid}
No extra headers required.
"""
def download_message(message_id):
    response = make_mesh_request("inbox/%s" % (message_id,))
    logging.info("%s %s",response.status_code,response.text)

"""
Make a GET request to retrieve a subsequent chunks of a message from the inbox.
Uses the url /inbox/{messageid}/{chunknumber}
No extra headers required.
"""
def download_message_chunk(message_id,chunk_number):
    response = make_mesh_request("inbox/%s/%s" % (message_id,chunk_number))
    logging.info("%s %s",response.status_code,response.text)

"""
Make a PUT request to acknowledge downloading a message and remove it from the inbox.
Uses the url /inbox/{messageid}/status/acknowledged
No extra headers required.
"""
def acknowledge_message(message_id):
    response = make_mesh_request("inbox/%s/status/acknowledged" % (message_id,),do_put=True)
    logging.info("%s %s",response.status_code,response.text)

def test():
    try:
        authenticate()
    except MeshException as x:
        logging.error("Failed:%s %s" % (x.code,x))
    try:
        check_inbox()
    except MeshException as x:
        logging.error("Failed:%s %s" % (x.code,x))
    try:
        check_inbox_count()
    except MeshException as x:
        logging.error("Failed:%s %s" % (x.code,x))
    try:
        download_message(1)
    except MeshException as x:
        logging.error("Failed:%s %s" % (x.code,x))
    try:
        download_message_chunk(1,1)
    except MeshException as x:
        logging.error("Failed:%s %s" % (x.code,x))
    try:
        acknowledge_message(1)
    except MeshException as x:
        logging.error("Failed:%s %s" % (x.code,x))

if __name__=="__main__":
    logging.basicConfig(level="DEBUG")
    test()