Authentication & Authorization
This section covers how the ZipRight back-end application authenticates and authorizes itself to interact with the EPC platforms Partner facing API.
Authentication/Authorization Implementation in ZipRight
ZipRight has a utility class in its back-end that handles maintaining a valid EPC oAuth token at the application level - caching the token for a little under 7200 seconds (as a buffer).
Whenever the cached token is needed for an API request to the EPC platform, it is accessed from the single instance of the token_manager
object shared across the back-end:
...
""" Token Manager module for oAuth token re-use """
class token_manager:
def __init__(self, grant_type, scope):
self.url = EPC_BASE_URL + EPC_OAUTH_URL
self.client_id = EPC_CLIENT_ID
self.client_secret = EPC_CLIENT_SECRET
self.grant_type = grant_type
self.scope = scope
self.access_token = ""
self.expiry = None
self.create_token()
""" Internal function for creating EPC oAuth token """
def _create_token(self):
if self.client_secret is not None and self.client_id is not None:
body = {
'grant_type': self.grant_type,
'client_id': self.client_id,
'client_secret': self.client_secret,
'scope': self.scope
}
content_type = MIMETYPES.get('FORM')
response = post(url=self.url, content_type=content_type, body=body)
if response is not None:
response_obj = response.json()
self.access_token = response_obj.get('access_token')
# Set expiry to a little under 2 hrs - 7200 seconds
self.expiry = time.time() + 7190
return True
return False
""" Public function for retrieving cached EPC oAuth token """
def get_token(self) -> str:
current_time = time.time()
if current_time >= self.expiry:
self._create_token()
return self.access_token
oAuth Access Token Life-time
As mentioned in Getting Authorized, EPC oAuth tokens expire in two hours - or
7200
seconds - which is why the token cache expiry time in the snippet above is set to slightly under that time.
A single instance of the token_manager
is initialized within the root level back-end application instance:
...
""" Initializing the back-end application """
def create_app():
# Application Factory
app = connexion.FlaskApp(
__name__,
specification_dir="specifications/"
)
setup_token_manager(app)
setup_routes(app)
setup_request_tracker(app)
add_cors_support(app)
return app
""" Initializing the token manager"""
def setup_token_manager(app):
# Flask App context is encapsulated with Connexion app object
app.app.token_manager = token_manager(
grant_type='client_credentials',
scope='pc pcapi'
)
...
The token_manager
is then utilized whenever the cached oAuth token is needed for an API call to the EPC platform. For example - when retrieving origin information from EPC:
...
"""" Internal function to pull down Origin information """
def _get_origin(origin_id, partner_access_token) -> dict:
# Ask token_manager for current oAuth token
access_token = app.token_manager.get_token()
url = EPC_BASE_URL + EPC_ORIGINS_URL + origin_id
headers = {
'Authorization': 'Bearer' + ' ' + access_token,
'X-Elli-PAT': partner_access_token
}
response = get(url=url, headers=headers)
return response
...
Updated over 4 years ago