diff --git a/cspell.config.yaml b/cspell.config.yaml index df1d75a..43d7b30 100644 --- a/cspell.config.yaml +++ b/cspell.config.yaml @@ -11,6 +11,7 @@ ignorePaths: - "**/*.docx" - "**/*.png" - "**/*.emf" + - "**/*.py" words: - ACA - ALPN diff --git a/examples/python_gateway.py b/examples/python_gateway.py new file mode 100644 index 0000000..20f1267 --- /dev/null +++ b/examples/python_gateway.py @@ -0,0 +1,152 @@ +"""FDSH gateway class.""" + +from urllib.parse import urljoin + +import logging +import requests +import ssl +import uuid + +from requests.adapters import HTTPAdapter + + +class HubGateway: + + def __init__( + self, + base_url, + token_path, + client_id, + client_secret, + client_cert_path, + client_key_path, + resolve=None, + education_enrollment_path="/api/v1/education-enrollments", + ): + self.base_url = base_url + self.token_path = token_path + self.client_id = client_id + self.client_secret = client_secret + self.resolve = self._parse_resolve(resolve) + self.education_enrollment_path = education_enrollment_path + + self._token = None + self._session = requests.Session() + self._session.cert = (client_cert_path, client_key_path) + self._session.mount("https://", TLS1_2_Adapter(resolve_dict=self.resolve)) + + def get_education_enrollment_v1(self, payload): + """Get enrollment data from FDSH NSC API. + + Payload is a dictionary passed inside the nscRequest object, e.g. + { + "personGivenName": "Neil", + "personSurName": "Martinsen-Burrell", + "asOfDate": "1900-01-01", + "termsAcceptedIndicator": True, + "personBirthDate": "1999-01-01", + } + """ + return self._post(self.education_enrollment_path, {"nscRequest": payload}) + + @property + def access_token(self): + """Return the access token if we have one, otherwise get one.""" + if self._token is not None: + return self._token + + # make a request to get an access token + uri = urljoin(self.base_url, self.token_path) + # print("Getting token from ", uri) + response = self._session.post( + uri, + data={ + "grant_type": "client_credentials", + "client_id": self.client_id, + "client_secret": self.client_secret, + }, + ) + # token is inside the response + self._token = self._handle_response(response)["access_token"] + return self._token + + @staticmethod + def _handle_response(response): + """Return the JSON contents of the response unless there was an error.""" + # print("Handling response: ", response.status_code, response.content) + if response.status_code >= 200 and response.status_code < 300: + return response.json() + + if response.status_code == 401: + # token might have expired, clear it + self._token = None + + response.raise_for_status() + + def _post(self, url_path, data): + """Post data in a JSON request to the specified path. + + The url_path is joined to self.base_url. `data` is a dictionary + sent in the body of the JSON request. + """ + uri = urljoin(self.base_url, url_path) + request = requests.Request( + url=uri, + method="POST", + headers={ + "Content-type": "application/json", + "Authorization": f"Bearer {self.access_token}", + "messageID": str(uuid.uuid4()), + }, + json=data, + ) + return self._execute(request) + + def _execute(self, request): + """Handle a request on our specially configured TLS connection.""" + return self._handle_response(self._session.send(self._session.prepare_request(request))) + + @staticmethod + def _parse_resolve(resolve_string): + """Parse a string like impl.hub.cms.gov:8443:127.0.0.1. + + Returns None if the argument is false-y. + """ + if not resolve_string: + return None + host, port, ip = resolve_string.split(":") + + return {host: {port: ip}} + + +class TLS1_2_Adapter(HTTPAdapter): + """Force TLS version 1.2""" + + def __init__(self, *args, resolve_dict={}, **kwargs): + """Override DNS resolution for one host. + + resolve_dict has a hostname as the key and the value is a mapping from + port number to an IP address. + """ + self.resolve_dict = resolve_dict + super().__init__(*args, **kwargs) + + def get_connection(self, *args, **kwargs): + conn = super().get_connection(*args, **kwargs) + + if conn.host in self.resolve_dict: + # Save original hostname for SSL/SNI & validation + conn.assert_hostname = conn.host + conn.server_hostname = conn.host + + # Redirect the actual network destination socket to the IP + conn.host = self.resolve_dict[conn.host][conn.port] + return conn + + def init_poolmanager(self, *args, **kwargs): + context = ssl.create_default_context() + # Restrict both min and max to TLS 1.2 to completely force it + context.minimum_version = ssl.TLSVersion.TLSv1_2 + context.maximum_version = ssl.TLSVersion.TLSv1_2 + kwargs["ssl_context"] = context + return super().init_poolmanager(*args, **kwargs) diff --git a/examples/usage_example.py b/examples/usage_example.py new file mode 100644 index 0000000..c5332df --- /dev/null +++ b/examples/usage_example.py @@ -0,0 +1,58 @@ +import json +import os +import sys + +import requests + +import python_gateway + + +def run_live_test(): + # get OAuth client info from the environment + client_id = os.getenv("OAUTH_CLIENT_ID") + client_secret = os.getenv("OAUTH_CLIENT_SECRET") + + if (client_id is None) or (client_secret is None): + print("Error: could not read OAuth credentials from the environment") + sys.exit(1) + + # URL information + base_url = "https://impl.hub.cms.gov/" + token_path = "/auth/oauth/v2/token" + education_path = "/mesh/imp1/NationalStudentClearinghouseService" + + gateway = python_gateway.HubGateway( + base_url, + token_path, + client_id, + client_secret, + "/tmp/client.crt", + "/tmp/client.key", + resolve="impl.hub.cms.gov:8443:127.0.0.1", + education_enrollment_path=education_path, + ) + + print("Gateway initialized.") + + payload = { + "personGivenName": "Neil", + "personSurName": "Martinsen-Burrell", + "personBirthDate": "1999-01-01", + "asOfDate": "1900-01-01", + "termsAcceptedIndicator": True, + } + + try: + print("Attempting to get education enrollment...") + result_dict = gateway.get_education_enrollment_v1(payload) + print("Success!") + print(json.dumps(result_dict, indent=2)) + except requests.exceptions.HTTPError as e: + print("HTTP Error:", e) + except: + print("Unexpected Error:") + raise + + +if __name__ == "__main__": + run_live_test() diff --git a/examples/usage_example.sh b/examples/usage_example.sh new file mode 100755 index 0000000..cc9847e --- /dev/null +++ b/examples/usage_example.sh @@ -0,0 +1,43 @@ +#!/bin/bash +# Hub usage example using `curl` and `jq` +# +# Assumes the tunneling setup from ../guides/ssh-tunneling-example.md +# where localhost:8443 is tunneled to impl.hub.cms.gov:443 +# +# client certificate is in /tmp/client.crt and its private key is in +# /tmp/client.key. OAuth ID and secret are in environment variables + +OAUTH_CLIENT_KEY=${OAUTH_CLIENT_KEY} +OAUTH_CLIENT_SECRET=${OAUTH_CLIENT_SECRET} + +# get access token from /auth/oauth/v2/token +# The encoded token is in the "access_token" property of the JSON response +echo "Getting access token..." +ACCESS_TOKEN=$(curl \ + --cert /tmp/client.crt \ + --key /tmp/client.key \ + --tlsv1.2 --tls-max 1.2 \ + --resolve "impl.hub.cms.gov:8443:127.0.0.1" \ + -X POST \ + https://impl.hub.cms.gov:8443/auth/oauth/v2/token \ + -H "Content-Type: application/x-www-form-urlencoded" \ + -d "grant_type=client_credentials&client_id=${OAUTH_CLIENT_KEY}&client_secret=${OAUTH_CLIENT_SECRET}" \ +| jq -r .access_token) + +echo "Calling NSC API..." +# call the NSC search API +curl -H "Authorization: Bearer ${ACCESS_TOKEN}" \ + --cert /tmp/client.crt \ + --key /tmp/client.key \ + --tlsv1.2 --tls-max 1.2 \ + --resolve "impl.hub.cms.gov:8443:127.0.0.1" \ + https://impl.hub.cms.gov:8443/mesh/imp1/NationalStudentClearinghouseService \ + -H "messageID: anything-seems-to-work-here" \ + --json '{"nscRequest": { + "personGivenName": "Neil", + "personSurName": "Martinsen-Burrell", + "asOfDate": "1900-01-01", + "termsAcceptedIndicator": true, + "personBirthDate": "1999-01-01" + } + }'