Added SSO to retrieve credentials for calling Cognito related queries

.devcontainer/docker-compose-dev.yml

@@ -16,11 +16,12 @@ services:
       - ../../backend.env
     volumes:
       - ../../:/portal
-      - ~/.aws:/.aws
+      - ~/.aws:/root/.aws
       - /var/run/docker.sock:/var/run/docker.sock
       - ~/.gitconfig:/.gitconfig
     command: /bin/sh -c "while sleep 1000; do :; done"
     network_mode: host
     environment:
-      AWS_SHARED_CREDENTIALS_FILE: /.aws/credentials
-      AWS_CONFIG_FILE: /.aws/config
+      AWS_SHARED_CREDENTIALS_FILE: /root/.aws/credentials
+      AWS_CONFIG_FILE: /root/.aws/config
+      AWS_PROFILE: cpac-webmaster

app.py

@@ -10,7 +10,6 @@
 from starlette.middleware.base import _StreamingResponse
 from typing import Awaitable, Callable
 
-
 app = FastAPI()
 auth = Cognito(
     region=COGNITO_REGION,

controllers/contract.py

@@ -48,7 +48,7 @@ async def post(self, item: PostItem, current_user: CognitoClaims = Depends(get_c
                 helpers=item.helpers,
                 num_additional_chairs=item.num_additional_chairs,
                 signer_email=current_user.email,  # TODO assert that emails are verified
-                signer_name=current_user.username,
+                signer_name=current_user.username,  # TODO signer_name should be the user's name, not username
                 artist_phone_number=item.artist_phone_number  # TODO this should be stored in AWS
                 )
         except NoApproverException:

controllers/me.py

@@ -72,7 +72,7 @@ async def patch(self, current_user: CognitoClaims = Depends(get_current_user)) -
 
     async def post(self, item: PostItem, current_user: CognitoClaims = Depends(get_current_user)) -> Response:  # type: ignore[no-any-unimported]
         try:
-            ret: bool = await MeManager().create_user(current_user.sub, str(item.vendor_type))
+            ret: bool = await MeManager().create_user(current_user.sub, current_user.username, str(item.vendor_type))
         except DuplicateKeyError:
             raise HTTPException(
                 status_code=status.HTTP_409_CONFLICT,

database/__init__.py

@@ -1,2 +1,3 @@
 from .users import UsersDB
-from .contracts import ContractsDB
+from .contracts import ContractsDB
+from .cognito import CognitoIdentityProviderWrapper

database/cognito.py

@@ -0,0 +1,49 @@
+# https://docs.aws.amazon.com/code-library/latest/ug/python_3_cognito-identity-provider_code_examples.html
+
+from botocore.exceptions import ClientError
+import logging
+import boto3
+from config.env import COGNITO_USERPOOL_ID
+from utilities.types import JSONDict
+from distutils.util import strtobool
+
+
+class CognitoUser:
+
+    def __init__(self, cognito_response_json: JSONDict) -> None:
+        assert cognito_response_json.get("UserAttributes")
+        for attribute_dict in cognito_response_json['UserAttributes']:
+            assert "Name" in attribute_dict
+            assert attribute_dict.get("Value")
+            match attribute_dict['Name']:
+                case 'sub':
+                    self.sub: str = attribute_dict['Value']
+                case 'email_verified':
+                    self.email_verified: bool = bool(strtobool(attribute_dict['Value']))
+                case 'email':
+                    self.email: str = attribute_dict['Value']
+        assert self.sub is not None
+        assert self.email_verified is not None
+        assert self.email is not None
+
+
+class CognitoIdentityProviderWrapper:
+    """Encapsulates Amazon Cognito actions"""
+    def __init__(self) -> None:
+        self.cognito_idp_client = boto3.client('cognito-idp')
+
+    def get_user(self, username: str) -> CognitoUser:
+        """
+        Gets a user in Cognito by it's username.
+
+        :return: user
+        """
+        try:
+            response: JSONDict = self.cognito_idp_client.admin_get_user(UserPoolId=COGNITO_USERPOOL_ID, Username=username)
+            assert type(response) is dict
+            return CognitoUser(response)
+        except ClientError as err:
+            logging.error(
+                "Couldn't list users for %s. Here's why: %s: %s", COGNITO_USERPOOL_ID,
+                err.response['Error']['Code'], err.response['Error']['Message'])
+            raise err

database/users.py

@@ -10,9 +10,10 @@
 class UsersDB(BaseDB):
 
     @classmethod
-    def _new_user(cls, _id: str, vendor_type: str) -> JSONDict:
+    def _new_user(cls, _id: str, username: str, vendor_type: str) -> JSONDict:
         return {
             "_id": _id,
+            "username": username,
             "contracts": [],
             "group": Groups.CUSTOMER,
             "vendor_type": vendor_type,
@@ -26,9 +27,8 @@ async def get_user(cls, uuid: str) -> Optional[MongoMappingType]:
         return result
 
     @classmethod
-    async def create_user(cls, uuid: str, vendor_type: str) -> bool:
-        query = cls._new_user(uuid, vendor_type)
-        # TODO catch error if user already exists
+    async def create_user(cls, uuid: str, username: str, vendor_type: str) -> bool:
+        query = cls._new_user(uuid, username, vendor_type)
         ret: pymongo.results.InsertOneResult = await cls.get_collection().insert_one(query)
         return ret.acknowledged
 

managers/contract.py

@@ -5,6 +5,7 @@
 from database import UsersDB
 from utilities import NoApproverException
 from typing import List
+from database import CognitoIdentityProviderWrapper
 
 
 class ContractManager():
@@ -28,11 +29,14 @@ async def create_contract(
         approver = await UsersDB.get_random_artist_reviewer()
         if approver is None:
             raise NoApproverException()
-        # TODO cannot do this to get approver. Need to grab from cognito DB
-        # approver_email = approver.get("email")
-        # approver_name = approver.get('name')
-        approver_email = "[email protected]"
-        approver_name = "test"
+
+        # Get the email and username from CognitoDB
+        assert "username" in approver
+        approver_name = approver["username"]
+        approver_cognito_data = CognitoIdentityProviderWrapper().get_user(approver_name)
+        approver_email = approver_cognito_data.email
+
+        # TODO the approver_name should be the user's name, not username
 
         assert type(approver_email) is str, "Expected a string for approver email"
         assert type(approver_name) is str, "Expected a string for approver name"

managers/me.py

@@ -3,5 +3,5 @@
 
 class MeManager():
 
-    async def create_user(self, user_id: str, vendor_type: str) -> bool:
-        return await UsersDB.create_user(user_id, vendor_type)
+    async def create_user(self, user_id: str, username: str, vendor_type: str) -> bool:
+        return await UsersDB.create_user(user_id, username, vendor_type)

mypy.ini

@@ -26,4 +26,10 @@ ignore_missing_imports = True
 ignore_missing_imports = True
 
 [mypy-motor.motor_asyncio]
+ignore_missing_imports = True
+
+[mypy-botocore.exceptions]
+ignore_missing_imports = True
+
+[mypy-boto3]
 ignore_missing_imports = True

requirements.txt

@@ -3,6 +3,8 @@ anyio==3.7.1
 asyncio==3.4.3
 attrs==23.1.0
 blinker==1.6.2
+boto3==1.28.48
+botocore==1.31.48
 certifi==2023.7.22
 cffi==1.15.1
 charset-normalizer==3.2.0
@@ -23,6 +25,7 @@ httpcore==0.17.3
 idna==3.4
 itsdangerous==2.1.2
 Jinja2==3.1.2
+jmespath==1.0.1
 jsonschema==4.18.4
 jsonschema-specifications==2023.7.1
 MarkupSafe==2.1.3
@@ -48,10 +51,11 @@ referencing==0.30.0
 requests==2.31.0
 rpds-py==0.9.2
 rsa==4.9
+s3transfer==0.6.2
 six==1.16.0
 sniffio==1.3.0
 tomli==2.0.1
 types-jsonschema==4.17.0.10
 typing_extensions==4.7.1
-urllib3==2.0.4
-Werkzeug==2.3.6
+urllib3==1.26.16
+Werkzeug==2.3.6