feat(auth) convert boto error's into staxapp errors

Author: ZXYmania

Pipfile

@@ -0,0 +1,28 @@
+[[source]]
+name = "pypi"
+url = "https://pypi.org/simple"
+verify_ssl = true
+
+[dev-packages]
+
+[packages]
+aws-requests-auth = "*"
+black = "*"
+boto3 = "*"
+isort = "*"
+jsonschema = "*"
+nose2 = "*"
+openapi-spec-validator = "*"
+prance = "*"
+pycodestyle = "*"
+pycrypto = "*"
+pylint = "*"
+pytest = "*"
+pytest-cov = "*"
+requests = "*"
+responses = "*"
+warrant = "*"
+PyJWT = "*"
+
+[requires]
+python_version = "3.7"

examples/test.py

@@ -0,0 +1,51 @@
+import os
+import sys
+import boto3
+
+from staxapp.config import Config
+from staxapp.openapi import StaxClient
+from staxapp.api import Api
+from staxapp.exceptions import ApiException
+
+sts = boto3.client("sts")
+
+security_account = '750975847145'
+api_token_name = 'dean-token'
+
+response = sts.assume_role(RoleArn=f"arn:aws:iam::{security_account}:role/{api_token_name}-access-role", RoleSessionName=f"{api_token_name}-ssm-role")
+assumed_ssm = boto3.client("ssm",
+aws_access_key_id=response["Credentials"]["AccessKeyId"],
+aws_secret_access_key=response["Credentials"]["SecretAccessKey"],
+aws_session_token=response["Credentials"]["SessionToken"]
+)
+api_token_access_key = assumed_ssm.get_parameter(Name=f"/stax/api-tokens/{api_token_name}/AccessKey", WithDecryption=True)
+api_token_secret_key = assumed_ssm.get_parameter(Name=f"/stax/api-tokens/{api_token_name}/SecretKey", WithDecryption=True)
+
+Config.access_key = api_token_access_key['Parameter']['Value']
+Config.secret_key = api_token_secret_key['Parameter']['Value']
+
+Config.access_key = 'fake'
+Config.secret_key = 'fake'
+
+
+# fake_client = StaxClient("fake")
+client = StaxClient("accounts")
+
+allAccounts = client.ReadAccounts()
+print(f'{len(allAccounts["Accounts"])}')
+print(client.ReadAccounts(limit=1, offset=0))
+print(client.ReadAccounts(account_id="9fc4fd2e-1b4a-49b9-a341-d7ee77ea132d"))
+print(client.ReadAccounts(filter="ERROR", account_id="9fc4fd2e-1b4a-49b9-a341-d7ee77ea132d"))
+
+
+client = StaxClient("workloads")
+# client.FakeMethod()
+# print(client.ReadCatalogueItems())
+response = client.ReadCatalogueItems(catalogue_id='9c4fc016-5221-460d-8bf8-4104178e9e10')
+print(client.ReadCatalogueVersion(version_id='545489ae-c090-45cd-9322-42f9b2ed7b6a', catalogue_id='9c4fc016-5221-460d-8bf8-4104178e9e10'))
+print(client.ReadCatalogueVersion(version_id='d58ad318-fa36-4310-9766-e7f5e4a34f8d', include_parameters=False, catalogue_id='f13dd683-4aa6-4b88-abc8-ad58a7ee04f9'))
+
+# print(client.DeleteCatalogueVersion(catalogue_id='fake'))
+print(client.ReadCatalogueVersion(version_id='d58ad318-fa36-4310-9766-e7f5e4a34f8d', include_parameters=False))
+
+# print(fake_client.ReadAccounts(limit=1, offset=0))

staxapp/auth.py

@@ -11,6 +11,7 @@
 from warrant import AWSSRP, Cognito
 
 from staxapp.config import Config as JumaConfig
+from staxapp.exceptions import InvalidCredentialsException
 
 
 class StaxAuth:
@@ -23,6 +24,11 @@ def __init__(self, config_branch):
         self.aws_region = config.get(config_branch).get("region")
 
     def requests_auth(self, username, password):
+        if username is None:
+            raise InvalidCredentialsException("Please provide an Access Key to your config")
+        if password is None:
+            raise InvalidCredentialsException("Please provide a Secret Key to your config")
+
         id_token = self.id_token_from_cognito(username, password)
         id_creds = self.sts_from_cognito_identity_pool(id_token)
         auth = self.sigv4_signed_auth_headers(id_creds)
@@ -47,7 +53,14 @@ def id_token_from_cognito(self, username=None, password=None):
                 client_id=self.client_id,
                 client=client,
             )
-            tokens = aws.authenticate_user()
+            try:
+                tokens = aws.authenticate_user()
+            except client.exceptions.NotAuthorizedException as e:
+                logging.error(e)
+                raise InvalidCredentialsException(message=str(e), detail="Please check your Secret Key is correct")
+            except client.exceptions.UserNotFoundException as e:
+                raise InvalidCredentialsException(message=str(e), detail="Please check your Access Key, that you have created your Api Token and that you are using the right STAX REGION")
+
             # logging.debug(f"TOKEN: {tokens}")
             token = tokens["AuthenticationResult"]["IdToken"]
         else:

staxapp/exceptions.py

@@ -26,3 +26,14 @@ class ValidationException(Exception):
     def __init__(self, message):
         # logging.info(f"VALIDATE: {message}")
         self.message = message
+
+
+class InvalidCredentialsException(Exception):
+    def __init__(self, message, detail=""):
+        prefix = f"InvalidCredentialsException: "
+        if detail:
+            prefix = f"{prefix}{detail} - "
+        self.message = f"{prefix}{message}"
+
+    def __str__(self):
+        return self.message

tests/test_auth.py

@@ -15,7 +15,7 @@
 from botocore.stub import Stubber
 
 from staxapp.auth import StaxAuth
-
+from staxapp.exceptions import InvalidCredentialsException
 
 class StaxAuthTests(unittest.TestCase):
     """
@@ -63,6 +63,20 @@ def testCreds(self):
         self.assertIn("Credentials", creds)
         self.assertTrue(creds.get("IdentityId").startswith("ap-southeast-2"))
 
+    def testCredentialErrors(self):
+      """
+      Test that errors are thrown when keys are invalid
+      """
+      sa = StaxAuth("ApiAuth")
+      # Test with no username
+      with self.assertRaises(InvalidCredentialsException):
+        sa.requests_auth(username=None, password='valid')
+      # Test with no username
+      with self.assertRaises(InvalidCredentialsException):
+        sa.requests_auth(username='valid', password=None)
+      # Test with invalid username password
+      # Todo
+
     def stub_cognito_creds(self, token: str):
         sa = StaxAuth("ApiAuth")