BREAKING CHANGE : Catalia daac delivery #612

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open

wphyojpl wants to merge 33 commits into develop from catalia-daac-delivery

0 ...bda_functions/cumulus_wrapper/__init__.py → ...bda_functions/catalya_uds_api/__init__.py

File renamed without changes.

cumulus_lambda_functions/catalya_uds_api/auth_admin_api.py

-Original file line number
+Diff line change
@@ -0,0 +1,204 @@
+    from typing import Union
+    from cumulus_lambda_functions.daac_archiver.catalia_auth_db import CataliaAuthDb
+    from cumulus_lambda_functions.lib.lambda_logger_generator import LambdaLoggerGenerator
+    from cumulus_lambda_functions.uds_api.fast_api_utils import FastApiUtils
+    from cumulus_lambda_functions.uds_api.web_service_constants import WebServiceConstants
+    from fastapi import APIRouter, HTTPException, Request, Response
+    LOGGER = LambdaLoggerGenerator.get_logger(__name__, LambdaLoggerGenerator.get_level_from_env())
+    import json
+    import os
+    from pydantic import BaseModel
+    from cumulus_lambda_functions.lib.authorization.uds_authorizer_abstract import UDSAuthorizorAbstract
+    from cumulus_lambda_functions.lib.authorization.uds_authorizer_factory import UDSAuthorizerFactory
+    from mdps_ds_lib.lib.utils.json_validator import JsonValidator
+    from cumulus_lambda_functions.lib.lambda_logger_generator import LambdaLoggerGenerator
+    LOGGER = LambdaLoggerGenerator.get_logger(__name__, LambdaLoggerGenerator.get_level_from_env())
+    class AuthDeleteModel(BaseModel):
+        source: str
+        target: str
+        group_name: str
+    delete_schema = {
+        'type': 'object',
+        'required': ['source', 'target', 'group_name'],
+        'properties': {
+            'source': {'type': 'string'},
+            'target': {'type': 'string'},
+            'group_name': {'type': 'string'},
+        }
+    }
+    class AuthListModel(BaseModel):
+        group_name: list[str]
+    list_schema = {
+        'type': 'object',
+        'properties': {
+            'tenant': {'type': 'string'},
+            'venue': {'type': 'string'},
+            'group_names': {
+                'type': 'array',
+                'items': {'type': 'string'},
+                'minItems': 1,
+            },
+        }
+    }
+    class AuthAddModel(BaseModel):
+        source: str
+        target: str
+        group_name: str
+        access: bool
+    add_schema = {
+        'type': 'object',
+        'required': ['source', 'target', 'group_name', 'access'],
+        'properties': {
+            'source': {'type': 'string'},
+            'target': {'type': 'string'},
+            'group_name': {'type': 'string'},
+            'access': {'type': 'boolean'},
+        }
+    }
+    class AuthCrud:
+        def __init__(self, authorization_info, request_body):
+            required_env = ['ADMIN_COMMA_SEP_GROUPS', 'CATALYA_DB_NAME']
+            if not all([k in os.environ for k in required_env]):
+                raise EnvironmentError(f'one or more missing env: {required_env}')
+            self.__request_body = request_body
+            self.__authorization_info = authorization_info
+            self.__admin_groups = [k.strip() for k in os.getenv('ADMIN_COMMA_SEP_GROUPS').split(',')]
+            self.__cad = CataliaAuthDb(os.getenv('CATALYA_DB_NAME'))
+        def is_admin(self):
+            belonged_admin_groups = list(set(self.__admin_groups) & set(self.__authorization_info['ldap_groups']))
+            if len(belonged_admin_groups) < 1:
+                LOGGER.warn(f'unauthorized attempt to admin function: {self.__authorization_info}')
+                return {
+                    'statusCode': 403,
+                    'body': {'message': f'user is not in admin groups: {self.__admin_groups}'}
+                }
+            return {
+                    'statusCode': 200,
+                    'body': {}
+                }
+        def list_all_record(self):
+            return {
+                    'statusCode': 501,
+                    'body': {'message': 'Not Implemented Yet'}
+                }
+            # return {
+            #         'statusCode': 200,
+            #         'body': all_records
+            #     }
+        def add_new_record(self):
+            body_validator_result = JsonValidator(add_schema).validate(self.__request_body)
+            if body_validator_result is not None:
+                LOGGER.error(f'invalid add body: {body_validator_result}. request_body: {self.__request_body}')
+                return {
+                    'statusCode': 500,
+                    'body': f'invalid add body: {body_validator_result}. request_body: {self.__request_body}'
+                }
+            self.__cad.add(self.__request_body['group_name'], self.__request_body['source'], self.__request_body['target'], self.__request_body['access'])
+            return {
+                'statusCode': 200,
+                'body': {'message': 'inserted'}
+            }
+        def delete_record(self):
+            body_validator_result = JsonValidator(delete_schema).validate(self.__request_body)
+            if body_validator_result is not None:
+                LOGGER.error(f'invalid delete body: {body_validator_result}. request_body: {self.__request_body}')
+                return {
+                    'statusCode': 500,
+                    'body': f'invalid delete body: {body_validator_result}. request_body: {self.__request_body}'
+                }
+            self.__cad.delete(self.__request_body['group_name'], self.__request_body['source'], self.__request_body['target'])
+            return {
+                'statusCode': 200,
+                'body': {'message': 'deleted'}
+            }
+    router = APIRouter(
+        prefix=f'/{WebServiceConstants.ADMIN}/auth',
+        tags=["Admin Records CRUD (Admins-Only)"],
+        responses={404: {"description": "Not found"}},
+    )
+    @router.delete("")
+    @router.delete("/")
+    async def delete_auth_mapping(request: Request, delete_body: AuthDeleteModel):
+        """
+        Deleting one authorization mapping
+        """
+        LOGGER.debug(f'started delete_auth_mapping')
+        auth_info = FastApiUtils.get_authorization_info(request)
+        auth_crud = AuthCrud(auth_info, delete_body.model_dump())
+        is_admin_result = auth_crud.is_admin()
+        if is_admin_result['statusCode'] != 200:
+            raise HTTPException(status_code=is_admin_result['statusCode'], detail=is_admin_result['body'])
+        delete_result = auth_crud.delete_record()
+        if delete_result['statusCode'] == 200:
+            return delete_result['body']
+        raise HTTPException(status_code=delete_result['statusCode'], detail=delete_result['body'])
+    @router.post("")
+    @router.post("/")
+    async def add_auth_mapping(request: Request, new_body: AuthAddModel):
+        """
+        Adding a new Authorization mapping
+        """
+        LOGGER.debug(f'started add_auth_mapping. sss {new_body.model_dump()}')
+        auth_info = FastApiUtils.get_authorization_info(request)
+        auth_crud = AuthCrud(auth_info, new_body.model_dump())
+        is_admin_result = auth_crud.is_admin()
+        if is_admin_result['statusCode'] != 200:
+            raise HTTPException(status_code=is_admin_result['statusCode'], detail=is_admin_result['body'])
+        add_result = auth_crud.add_new_record()
+        if add_result['statusCode'] == 200:
+            return add_result['body']
+        raise HTTPException(status_code=add_result['statusCode'], detail=add_result['body'])
+    @router.get("")
+    @router.get("/")
+    async def list_auth_mappings(request: Request, tenant: Union[str, None]=None, venue: Union[str, None]=None, group_names: Union[str, None]=None):
+        """
+        Listing all exsiting Authorization Mapping.
+        """
+        LOGGER.debug(f'started list_auth_mappings')
+        auth_info = FastApiUtils.get_authorization_info(request)
+        query_body = {
+            'tenant': tenant,
+            'venue': venue,
+            'ldap_group_names': group_names if group_names is None else [k.strip() for k in group_names.split(',')],
+        }
+        auth_crud = AuthCrud(auth_info, query_body)
+        is_admin_result = auth_crud.is_admin()
+        if is_admin_result['statusCode'] != 200:
+            raise HTTPException(status_code=is_admin_result['statusCode'], detail=is_admin_result['body'])
+        query_result = auth_crud.list_all_record()
+        if query_result['statusCode'] == 200:
+            return query_result['body']
+        raise HTTPException(status_code=query_result['statusCode'], detail=query_result['body'])

cumulus_lambda_functions/catalya_uds_api/granules_archive_api.py

-Original file line number
+Diff line change
@@ -0,0 +1,132 @@
+    import json
+    import os
+    from typing import Optional
+    from cumulus_lambda_functions.daac_archiver.catalia_auth_db import CataliaAuthDb
+    from cumulus_lambda_functions.daac_archiver.catalia_daac_handshakes_db import CataliaDaacHandshakesDb
+    from cumulus_lambda_functions.daac_archiver.daac_archiver_catalia import DaacArchiverCatalia
+    from cumulus_lambda_functions.lib.lambda_logger_generator import LambdaLoggerGenerator
+    from cumulus_lambda_functions.uds_api.web_service_constants import WebServiceConstants
+    from cumulus_lambda_functions.uds_api.fast_api_utils import FastApiUtils
+    from fastapi import APIRouter, HTTPException, Request
+    from pydantic import BaseModel
+    LOGGER = LambdaLoggerGenerator.get_logger(__name__, LambdaLoggerGenerator.get_level_from_env())
+    router = APIRouter(
+        prefix=f'/{WebServiceConstants.COLLECTIONS}',
+        tags=["Granules Archive CRUD API"],
+        responses={404: {"description": "Not found"}},
+    )
+    class ArchivingTypesModel(BaseModel):
+        data_type: str
+        file_extension: Optional[list[str]] = []
+    class DaacUpdateModel(BaseModel):
+        daac_collection_id: str
+        api_key: str
+        daac_provider: Optional[str] = None
+        daac_data_version: Optional[str] = None
+        daac_sns_topic_arn: Optional[str] = None
+        daac_role_arn: Optional[str] = None
+        daac_role_session_name: Optional[str] = None
+        archiving_types: Optional[list[ArchivingTypesModel]] = None
+    class InternalDDBConnector:
+        def __init__(self):
+            required_env = ['CATALYA_DAAC_AGREEMENT_DB_NAME', 'CATALYA_DB_NAME']
+            if not all([k in os.environ for k in required_env]):
+                raise EnvironmentError(f'one or more missing env: {required_env}')
+            self.cad = CataliaAuthDb(os.getenv('CATALYA_DB_NAME'))
+            self.cdhsd = CataliaDaacHandshakesDb(os.getenv('CATALYA_DAAC_AGREEMENT_DB_NAME'))
+            self.auth_info = {}
+            self.configured_daac_configs = []
+        def archive_methods_initiator(self, request, collection_id, daac_collection_id):
+            LOGGER.debug(f'started archive_methods_initiator.')
+            self.auth_info = FastApiUtils.get_authorization_info(request)
+            if daac_collection_id is None:
+                self.configured_daac_configs = self.cdhsd.search(collection_id)
+                configured_daac_ids = [k[self.cdhsd.target_project] for k in self.configured_daac_configs]
+            else:
+                configured_daac_ids = [daac_collection_id]
+            authorized_daacs = self.cad.get_authorized_daac_full(self.auth_info.get('ldap_groups'), collection_id, configured_daac_ids)
+            if len(authorized_daacs) < 1:
+                LOGGER.debug(f'user: {self.auth_info["username"]} is not authorized for {collection_id}')
+                raise HTTPException(status_code=403, detail=json.dumps({
+                    'message': 'not authorized to execute this action'
+                }))
+            return authorized_daacs
+    @router.post("/{collection_id}/{daac_collection_id}/archive")
+    @router.post("/{collection_id}/{daac_collection_id}/archive/")
+    async def add_daac_archive_config(request: Request, collection_id: str, daac_collection_id: str, new_body: DaacUpdateModel):
+        LOGGER.debug(f'started add_daac_archive_config. {new_body.model_dump()}')
+        i1 = InternalDDBConnector()
+        authorized_daacs = i1.archive_methods_initiator(request, collection_id, daac_collection_id)
+        authorized_ldaps = [k['userGroup'] for k in authorized_daacs]
+        b1 = new_body.model_dump()
+        try:
+            # def add(self, catalia_collection, daac_collection, api_key, provider, data_version, sns_topic_arn, role_arn, role_session_name, archiving_types, user, user_group):
+            i1.cdhsd.add(collection_id, daac_collection_id, b1['api_key'], b1['daac_provider'], b1['daac_data_version'],
+                         b1['daac_sns_topic_arn'], b1['daac_role_arn'], b1['daac_role_session_name'], b1['archiving_types'], i1.auth_info['username'], authorized_ldaps)
+        except Exception as e:
+            LOGGER.exception(f'error while add_daac_archive_config: {b1}')
+            raise HTTPException(status_code=500, detail=e)
+        return {'message': 'archive config added'}
+    @router.delete("/{collection_id}/{daac_collection_id}/archive")
+    @router.delete("/{collection_id}/{daac_collection_id}/archive/")
+    async def delete_daac_archive_config(request: Request, collection_id: str, daac_collection_id: str):
+        LOGGER.debug(f'started delete_daac_archive_config.')
+        i1 = InternalDDBConnector()
+        authorized_daacs = i1.archive_methods_initiator(request, collection_id, daac_collection_id)
+        try:
+            i1.cdhsd.delete(collection_id, daac_collection_id)
+        except Exception as e:
+            LOGGER.exception(f'error while delete_daac_archive_config: {collection_id}, {daac_collection_id}')
+            raise HTTPException(status_code=500, detail=e)
+        return {'message': 'archive config deleted'}
+    @router.get("/{collection_id}/{daac_collection_id}/archive")
+    @router.get("/{collection_id}/{daac_collection_id}/archive/")
+    async def get_daac_archive_config(request: Request, collection_id: str, daac_collection_id: str):
+        LOGGER.debug(f'started get_daac_archive_config.')
+        i1 = InternalDDBConnector()
+        authorized_daacs = i1.archive_methods_initiator(request, collection_id, daac_collection_id)
+        try:
+            result = i1.cdhsd.get_single(collection_id, daac_collection_id)
+        except Exception as e:
+            LOGGER.exception(f'error while get_daac_archive_config: {collection_id}, {daac_collection_id}')
+            raise HTTPException(status_code=500, detail=e)
+        return {'result': result}
+    @router.put("/{collection_id}/archive/{granule_id}")
+    @router.put("/{collection_id}/archive/{granule_id}/")
+    async def archive_single_granule(request: Request, collection_id: str, granule_id: str):
+        LOGGER.debug(f'started archive_single_granule.')
+        i1 = InternalDDBConnector()
+        authorized_daacs = i1.archive_methods_initiator(request, collection_id, None)
+        authorized_ldaps = set([k['userGroup'] for k in authorized_daacs])
+        authorized_configured_daac_configs = [k for k in i1.configured_daac_configs if k[i1.cdhsd.target_project] in authorized_ldaps]
+        dac = DaacArchiverCatalia()
+        dac.staged_s3_bucket = os.getenv('CATALYA_UDS_STAGING_BUCKET')
+        dac.daac_agreements = authorized_configured_daac_configs
+        dac.archive_granule(collection_id, granule_id)
+        return {'message': 'archive initiated'}
+    @router.put("/{collection_id}/archive")
+    @router.put("/{collection_id}/archive/")
+    async def archive_entire_collection(request: Request, collection_id: str):
+        LOGGER.debug(f'started archive_entire_collection.')
+        i1 = InternalDDBConnector()
+        authorized_daacs = i1.archive_methods_initiator(request, collection_id, None)
+        authorized_ldaps = set([k['userGroup'] for k in authorized_daacs])
+        authorized_configured_daac_configs = [k for k in i1.configured_daac_configs if k[i1.cdhsd.target_project] in authorized_ldaps]
+        dac = DaacArchiverCatalia()
+        dac.staged_s3_bucket = os.getenv('CATALYA_UDS_STAGING_BUCKET')
+        dac.daac_agreements = authorized_configured_daac_configs
+        dac.archive_collection(collection_id)  # TODO accept filtering mechanisms?
+        return {'message': 'archive initiated'}

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

BREAKING CHANGE : Catalia daac delivery #612

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

BREAKING CHANGE : Catalia daac delivery #612

Are you sure you want to change the base?

Uh oh!

BREAKING CHANGE : Catalia daac delivery #612

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!