diff --git a/server/admin_management/api/app/constants.py b/server/admin_management/api/app/constants.py new file mode 100644 index 000000000..6a6538559 --- /dev/null +++ b/server/admin_management/api/app/constants.py @@ -0,0 +1,9 @@ +from enum import Enum + + +class UserType(str, Enum): + IDIR = "I" + BCEID = "B" + + +COGNITO_USERNAME_KEY = "username" \ No newline at end of file diff --git a/server/admin_management/api/app/jwt_validation.py b/server/admin_management/api/app/jwt_validation.py index 609cfb1f8..ff51ea73b 100644 --- a/server/admin_management/api/app/jwt_validation.py +++ b/server/admin_management/api/app/jwt_validation.py @@ -12,6 +12,8 @@ get_user_pool_id, ) +from api.app.constants import COGNITO_USERNAME_KEY + JWT_GROUPS_KEY = "cognito:groups" JWT_CLIENT_ID_KEY = "client_id" @@ -199,3 +201,13 @@ def authorize(claims: dict = Depends(validate_token)) -> dict: def get_access_roles(claims: dict = Depends(authorize)): groups = claims[JWT_GROUPS_KEY] return groups + + +def get_request_cognito_user_id(claims: dict = Depends(authorize)): + # This is NOT user's name, display name or user ID. + # It is mapped to "cognito:username" (ID Token) and "username" (Access Token). + # It is the "cognito_user_id" column for fam_user table. + # Example value: idir_b5ecdb094dfb4149a6a8445a0mangled0@idir + cognito_username = claims[COGNITO_USERNAME_KEY] + LOGGER.debug(f"Current requester's cognito_username for API: {cognito_username}") + return cognito_username diff --git a/server/admin_management/api/app/main.py b/server/admin_management/api/app/main.py index beac205fc..dce8dd766 100644 --- a/server/admin_management/api/app/main.py +++ b/server/admin_management/api/app/main.py @@ -7,7 +7,7 @@ from mangum import Mangum from api.config.config import get_root_path, get_allow_origins -from api.app.routers import router_smoke_test +from api.app.routers import router_smoke_test, router_application_admin logConfigFile = os.path.join( @@ -19,7 +19,7 @@ apiPrefix = "" description = """ -Forest Access Management API used by the Forest Access Management application +Forest Access Management Admin Management API used by the Forest Access Management application to define admin access to forest applications. """ @@ -58,6 +58,11 @@ def custom_generate_unique_id(route: APIRouter): app.include_router( router_smoke_test.router, prefix=apiPrefix + "/smoke_test", tags=["Smoke Test"] ) +app.include_router( + router_application_admin.router, + prefix=apiPrefix + "/application_admin", + tags=["FAM Application Admin"], +) @app.get("/", include_in_schema=False, tags=["docs"]) diff --git a/server/admin_management/api/app/repositories/__init__.py b/server/admin_management/api/app/repositories/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/server/admin_management/api/app/repositories/application_admin_repository.py b/server/admin_management/api/app/repositories/application_admin_repository.py new file mode 100644 index 000000000..455af3d69 --- /dev/null +++ b/server/admin_management/api/app/repositories/application_admin_repository.py @@ -0,0 +1,78 @@ +import logging +from sqlalchemy.orm import Session +from typing import List + +from api.app.models import model as models + + +LOGGER = logging.getLogger(__name__) + + +class ApplicationAdminRepository: + def __init__(self, db: Session): + self.db = db + + def get_application_admin_by_app_and_user_id( + self, application_id: int, user_id: int + ) -> models.FamApplicationAdmin: + return ( + self.db.query(models.FamApplicationAdmin) + .filter( + models.FamApplicationAdmin.application_id == application_id, + models.FamApplicationAdmin.user_id == user_id, + ) + .one_or_none() + ) + + def get_application_admin_by_id( + self, application_admin_id: int + ) -> models.FamApplicationAdmin: + return ( + self.db.query(models.FamApplicationAdmin) + .filter( + models.FamApplicationAdmin.application_admin_id == application_admin_id + ) + .one_or_none() + ) + + def get_application_admin_by_application_id( + self, application_id: int + ) -> List[models.FamApplicationAdmin]: + return ( + self.db.query(models.FamApplicationAdmin) + .filter( + models.FamApplicationAdmin.application_id == application_id + ) + .all() + ) + + def create_application_admin( + self, application_id: int, user_id: int, requester: str + ) -> models.FamApplicationAdmin: + new_fam_application_admin: models.FamApplicationAdmin = ( + models.FamApplicationAdmin( + **{ + "user_id": user_id, + "application_id": application_id, + "create_user": requester, + } + ) + ) + self.db.add(new_fam_application_admin) + self.db.flush() + self.db.refresh(new_fam_application_admin) + LOGGER.debug( + f"New FamApplicationAdmin added for {new_fam_application_admin.__dict__}" + ) + return new_fam_application_admin + + def delete_application_admin(self, application_admin_id: int): + record = ( + self.db.query(models.FamApplicationAdmin) + .filter( + models.FamApplicationAdmin.application_admin_id == application_admin_id + ) + .one() + ) + self.db.delete(record) + self.db.flush() diff --git a/server/admin_management/api/app/repositories/application_repository.py b/server/admin_management/api/app/repositories/application_repository.py new file mode 100644 index 000000000..50836ca74 --- /dev/null +++ b/server/admin_management/api/app/repositories/application_repository.py @@ -0,0 +1,19 @@ +import logging +from sqlalchemy.orm import Session + +from api.app.models import model as models + + +LOGGER = logging.getLogger(__name__) + + +class ApplicationRepository: + def __init__(self, db: Session): + self.db = db + + def get_application(self, application_id: int) -> models.FamApplication: + return ( + self.db.query(models.FamApplication) + .filter(models.FamApplication.application_id == application_id) + .one_or_none() + ) diff --git a/server/admin_management/api/app/repositories/user_repository.py b/server/admin_management/api/app/repositories/user_repository.py new file mode 100644 index 000000000..746cc653e --- /dev/null +++ b/server/admin_management/api/app/repositories/user_repository.py @@ -0,0 +1,45 @@ +import logging +from sqlalchemy.orm import Session + +from api.app.models import model as models +from api.app import schemas + + +LOGGER = logging.getLogger(__name__) + + +class UserRepository: + def __init__(self, db: Session): + self.db = db + + def get_user_by_domain_and_name( + self, user_type_code: str, user_name: str + ) -> models.FamUser: + fam_user: models.FamUser = ( + self.db.query(models.FamUser) + .filter( + models.FamUser.user_type_code == user_type_code, + models.FamUser.user_name.ilike(user_name), + ) + .one_or_none() + ) + LOGGER.debug( + f"fam_user {str(fam_user.user_id) + ' found' if fam_user else 'not found'}." + ) + return fam_user + + def get_user_by_cognito_user_id(self, cognito_user_id: str) -> models.FamUser: + return ( + self.db.query(models.FamUser) + .filter(models.FamUser.cognito_user_id == cognito_user_id) + .one_or_none() + ) + + def create_user(self, fam_user: schemas.FamUser) -> models.FamUser: + LOGGER.debug(f"Creating fam user: {fam_user}") + + fam_user_dict = fam_user.model_dump() + db_item = models.FamUser(**fam_user_dict) + self.db.add(db_item) + self.db.flush() + return db_item diff --git a/server/admin_management/api/app/routers/router_application_admin.py b/server/admin_management/api/app/routers/router_application_admin.py new file mode 100644 index 000000000..ddd8444a8 --- /dev/null +++ b/server/admin_management/api/app/routers/router_application_admin.py @@ -0,0 +1,164 @@ +import logging +from fastapi import APIRouter, Depends, Request, Response, HTTPException +from sqlalchemy.orm import Session +from typing import List + + +from api.app.models import model as models +from api.app.routers.router_guards import ( + get_current_requester, + authorize_by_fam_admin, + enforce_self_grant_guard, + validate_param_application_admin_id, + validate_param_application_id, +) +from api.app import database, jwt_validation, schemas +from api.app.schemas import Requester +from api.app.services.application_admin_service import ApplicationAdminService +from api.app.services.user_service import UserService +from api.app.services.application_service import ApplicationService +from api.app.utils.audit_util import AuditEventLog, AuditEventOutcome, AuditEventType + +LOGGER = logging.getLogger(__name__) + +router = APIRouter() + + +@router.post( + "", + response_model=schemas.FamAppAdminGet, + dependencies=[ + Depends(authorize_by_fam_admin), + Depends(enforce_self_grant_guard), + Depends(validate_param_application_id), + ], +) +def create_application_admin( + application_admin_request: schemas.FamAppAdminCreate, + request: Request, + db: Session = Depends(database.get_db), + token_claims: dict = Depends(jwt_validation.authorize), + requester: Requester = Depends(get_current_requester), +): + + LOGGER.debug( + f"Executing 'create_application_admin' " + f"with request: {application_admin_request}, requestor: {token_claims}" + ) + + audit_event_log = AuditEventLog( + request=request, + event_type=AuditEventType.CREATE_APPLICATION_ADMIN_ACCESS, + event_outcome=AuditEventOutcome.SUCCESS, + ) + + try: + application_admin_service = ApplicationAdminService(db) + application_service = ApplicationService(db) + user_service = UserService(db) + + audit_event_log.requesting_user = user_service.get_user_by_cognito_user_id( + requester.cognito_user_id + ) + audit_event_log.application = application_service.get_application( + application_admin_request.application_id + ) + audit_event_log.target_user = user_service.get_user_by_domain_and_name( + application_admin_request.user_type_code, + application_admin_request.user_name, + ) + + return application_admin_service.create_application_admin( + application_admin_request, requester.cognito_user_id + ) + + except Exception as e: + audit_event_log.event_outcome = AuditEventOutcome.FAIL + audit_event_log.exception = e + raise e + + finally: + if audit_event_log.target_user is None: + audit_event_log.target_user = models.FamUser( + user_type_code=application_admin_request.user_type_code, + user_name=application_admin_request.user_name, + user_guid="unknown", + cognito_user_id="unknown", + ) + + audit_event_log.log_event() + + +@router.delete( + "/{application_admin_id}", + response_class=Response, + dependencies=[ + Depends(authorize_by_fam_admin), + Depends(enforce_self_grant_guard), + Depends(validate_param_application_admin_id), + ], +) +def delete_application_admin( + application_admin_id: int, + request: Request, + db: Session = Depends(database.get_db), + requester: Requester = Depends(get_current_requester), +): + LOGGER.debug( + f"Executing 'delete_application_admin' with request: {application_admin_id}" + ) + + audit_event_log = AuditEventLog( + request=request, + event_type=AuditEventType.REMOVE_APPLICATION_ADMIN_ACCESS, + event_outcome=AuditEventOutcome.SUCCESS, + ) + + try: + application_admin_service = ApplicationAdminService(db) + user_service = UserService(db) + + application_admin = application_admin_service.get_application_admin_by_id( + application_admin_id + ) + audit_event_log.requesting_user = user_service.get_user_by_cognito_user_id( + requester.cognito_user_id + ) + audit_event_log.application = application_admin.application + audit_event_log.target_user = application_admin.user + + return application_admin_service.delete_application_admin(application_admin_id) + + except Exception as e: + audit_event_log.event_outcome = AuditEventOutcome.FAIL + audit_event_log.exception = e + raise e + + finally: + audit_event_log.log_event() + + +@router.get( + "/{application_id}/admins", + response_model=List[schemas.FamAppAdminGet], + status_code=200, + dependencies=[Depends(authorize_by_fam_admin)], +) +def get_application_admin_by_applicationid( + application_id: int, + db: Session = Depends(database.get_db), +): + LOGGER.debug( + f"Loading application admin access for application_id: {application_id}" + ) + application_admin_service = ApplicationAdminService(db) + application_admin_access = ( + application_admin_service.get_application_admin_by_application_id( + application_id + ) + ) + LOGGER.debug( + f"Finished loading application admin access for application - # of results = {len(application_admin_access)}" + ) + + return application_admin_access diff --git a/server/admin_management/api/app/routers/router_guards.py b/server/admin_management/api/app/routers/router_guards.py index 3b4eafccd..a2283e03e 100644 --- a/server/admin_management/api/app/routers/router_guards.py +++ b/server/admin_management/api/app/routers/router_guards.py @@ -1,17 +1,43 @@ import logging from http import HTTPStatus -from fastapi import Depends, HTTPException +from fastapi import Depends, HTTPException, Request +from sqlalchemy.orm import Session +from typing import Union +import json +from api.app import database from api.app.jwt_validation import ( ERROR_PERMISSION_REQUIRED, get_access_roles, + get_request_cognito_user_id, validate_token, ) - +from api.app.schemas import Requester, TargetUser, FamAppAdminCreate +from api.app.models.model import FamUser +from api.app.services.application_admin_service import ApplicationAdminService +from api.app.services.user_service import UserService +from api.app.services.application_service import ApplicationService LOGGER = logging.getLogger(__name__) +ERROR_SELF_GRANT_PROHIBITED = "self_grant_prohibited" +ERROR_INVALID_APPLICATION_ID = "invalid_application_id" +ERROR_INVALID_ROLE_ID = "invalid_role_id" +ERROR_REQUESTER_NOT_EXISTS = "requester_not_exists" +ERROR_EXTERNAL_USER_ACTION_PROHIBITED = "external_user_action_prohibited" +ERROR_INVALID_APPLICATION_ADMIN_ID = "invalid_application_admin_id" + + +no_requester_exception = HTTPException( + status_code=HTTPStatus.FORBIDDEN, # 403 + detail={ + "code": ERROR_REQUESTER_NOT_EXISTS, + "description": "Requester does not exist, action is not allowed", + }, +) + + def authorize_by_fam_admin(claims: dict = Depends(validate_token)): required_role = "FAM_ACCESS_ADMIN" access_roles = get_access_roles(claims) @@ -25,3 +51,124 @@ def authorize_by_fam_admin(claims: dict = Depends(validate_token)): }, headers={"WWW-Authenticate": "Bearer"}, ) + + +async def get_current_requester( + request_cognito_user_id: str = Depends(get_request_cognito_user_id), + access_roles=Depends(get_access_roles), + db: Session = Depends(database.get_db), +): + user_service = UserService(db) + fam_user: FamUser = user_service.get_user_by_cognito_user_id( + request_cognito_user_id + ) + if fam_user is None: + raise no_requester_exception + + requester = Requester.model_validate(fam_user) + requester.access_roles = access_roles + LOGGER.debug(f"Current request user (requester): {requester}") + return requester + + +# Note!! +# currently to take care of different scenarios (id or fields needed in path/param/body) +# to find target user, will only consider request "path_params" and for "body"(json) for PUT/POST. +# For now, only consider known cases ("router_application_admin.py" endpoints that need this). +# Specifically: "user_role_xref_id" and "user_name/user_type_code". +# Very likely in future might have "cognito_user_id" case. +async def get_target_user_from_id( + request: Request, db: Session = Depends(database.get_db) +) -> Union[TargetUser, None]: + """ + This is used as FastAPI sub-dependency to find target_user for guard purpose. + For requester, use "get_current_requester()" above. + """ + # from path_param - application_admin_id, when remove admin access for a user + user_service = UserService(db) + if "application_admin_id" in request.path_params: + application_admin_service = ApplicationAdminService(db) + application_admin = application_admin_service.get_application_admin_by_id( + request.path_params["application_admin_id"] + ) + return ( + TargetUser.model_validate(application_admin.user) + if application_admin is not None + else None + ) + else: + # from body - {user_name/user_type_code}, when grant admin access + try: + rbody = await request.json() + user = user_service.get_user_by_domain_and_name( + rbody["user_type_code"], + rbody["user_name"], + ) + return TargetUser.model_validate(user) if user is not None else None + except json.JSONDecodeError: + return None + + +async def enforce_self_grant_guard( + requester: Requester = Depends(get_current_requester), + target_user: Union[TargetUser, None] = Depends(get_target_user_from_id), +): + """ + Verify logged on admin (requester): + Self granting/removing privilege currently isn't allowed. + """ + LOGGER.debug(f"enforce_self_grant_guard: requester - {requester}") + LOGGER.debug(f"enforce_self_grant_guard: target_user - {target_user}") + if target_user is not None: + is_same_user_name = requester.user_name == target_user.user_name + is_same_user_type_code = requester.user_type_code == target_user.user_type_code + + if is_same_user_name and is_same_user_type_code: + LOGGER.debug( + f"User '{requester.user_name}' should not " + f"grant/remove permission privilege to self." + ) + raise HTTPException( + status_code=HTTPStatus.FORBIDDEN, + detail={ + "code": ERROR_SELF_GRANT_PROHIBITED, + "description": "Altering permission privilege to self is not allowed", + }, + headers={"WWW-Authenticate": "Bearer"}, + ) + + +async def validate_param_application_admin_id( + application_admin_id: int, db: Session = Depends(database.get_db) +): + application_admin_service = ApplicationAdminService(db) + application_admin = application_admin_service.get_application_admin_by_id( + application_admin_id + ) + if not application_admin: + raise HTTPException( + status_code=HTTPStatus.BAD_REQUEST, + detail={ + "code": ERROR_INVALID_APPLICATION_ADMIN_ID, + "description": f"Application Admin ID {application_admin_id} not found", + }, + headers={"WWW-Authenticate": "Bearer"}, + ) + + +async def validate_param_application_id( + application_admin_request: FamAppAdminCreate, db: Session = Depends(database.get_db) +): + application_service = ApplicationService(db) + application = application_service.get_application( + application_admin_request.application_id + ) + if not application: + raise HTTPException( + status_code=HTTPStatus.BAD_REQUEST, + detail={ + "code": ERROR_INVALID_APPLICATION_ID, + "description": f"Application ID {application_admin_request.application_id} not found", + }, + headers={"WWW-Authenticate": "Bearer"}, + ) diff --git a/server/admin_management/api/app/schemas.py b/server/admin_management/api/app/schemas.py new file mode 100644 index 000000000..9a4bc6ef0 --- /dev/null +++ b/server/admin_management/api/app/schemas.py @@ -0,0 +1,69 @@ +import logging +from typing import List, Optional, Union + +from pydantic import StringConstraints, ConfigDict, BaseModel + +from . import constants as famConstants +from typing_extensions import Annotated + +LOGGER = logging.getLogger(__name__) + + +# -------------------------------------- Requester --------------------------------------- # +# Schema classes Requester and TargetUser are for backend system used and +# NOT intended as part of the request/respoinse body in the endpoint. Logged +# on user with jwt token is parsed into Requester (before route handler). +# Same as other Schema classes, it can be transformed from db model. +class Requester(BaseModel): + """ + Class holding information for user who access FAM system after authenticated. + """ + + # cognito_user_id => Cognito OIDC access token maps this to: username (ID token => "custom:idp_name" ) + cognito_user_id: Union[str, None] = None + user_name: Annotated[str, StringConstraints(max_length=20)] + # "B"(BCeID) or "I"(IDIR). It is the IDP provider. + user_type_code: Union[famConstants.UserType, None] = None + access_roles: Union[ + List[Annotated[str, StringConstraints(max_length=50)]], None + ] = None + + model_config = ConfigDict(from_attributes=True) + + +class TargetUser(Requester): + pass + + +# -------------------------------------- FAM User --------------------------------------- # +class FamUser(BaseModel): + user_type_code: famConstants.UserType + cognito_user_id: Optional[ + Annotated[str, StringConstraints(max_length=100)] + ] = None # temporarily optional + user_name: Annotated[str, StringConstraints(max_length=20)] + user_guid: Optional[Annotated[str, StringConstraints(max_length=32)]] = None + create_user: Annotated[str, StringConstraints(max_length=60)] + update_user: Optional[Annotated[str, StringConstraints(max_length=60)]] = None + + model_config = ConfigDict(from_attributes=True) + + +# -------------------------------------- FAM Application Admin --------------------------------------- # +# Application Admin assignment with one application at a time for the user. +class FamAppAdminCreate(BaseModel): + user_name: Annotated[ + str, StringConstraints(min_length=3, max_length=20) + ] + user_type_code: famConstants.UserType + application_id: int + + model_config = ConfigDict(from_attributes=True) + + +class FamAppAdminGet(BaseModel): + application_admin_id: int + application_id: int + user_id: int + + model_config = ConfigDict(from_attributes=True) diff --git a/server/admin_management/api/app/services/__init__.py b/server/admin_management/api/app/services/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/server/admin_management/api/app/services/application_admin_service.py b/server/admin_management/api/app/services/application_admin_service.py new file mode 100644 index 000000000..1537d9838 --- /dev/null +++ b/server/admin_management/api/app/services/application_admin_service.py @@ -0,0 +1,83 @@ +import logging +from http import HTTPStatus +from sqlalchemy.orm import Session +from typing import List + +from api.app import constants as famConstants +from api.app import schemas +from api.app.services.application_service import ApplicationService +from api.app.services.user_service import UserService +from api.app.repositories.application_admin_repository import ApplicationAdminRepository + +from api.app.utils import utils + +LOGGER = logging.getLogger(__name__) + + +class ApplicationAdminService: + def __init__(self, db: Session): + self.application_admin_repo = ApplicationAdminRepository(db) + self.application_service = ApplicationService(db) + self.user_service = UserService(db) + + def get_application_admin_by_id( + self, application_admin_id: int + ) -> schemas.FamAppAdminGet: + return self.application_admin_repo.get_application_admin_by_id( + application_admin_id + ) + + def get_application_admin_by_application_id( + self, application_id: int + ) -> List[schemas.FamAppAdminGet]: + return self.application_admin_repo.get_application_admin_by_application_id( + application_id + ) + + def create_application_admin( + self, request: schemas.FamAppAdminCreate, requester: str + ) -> schemas.FamAppAdminGet: + # Request has information: user_name, user_type_code, application_id + LOGGER.debug( + f"Request for assigning an application admin to a user: {request}." + ) + + # Verify if user already exists or add a new user + fam_user = self.user_service.find_or_create( + request.user_type_code, request.user_name, requester + ) + + # Verify if user is admin already + fam_application_admin_user = ( + self.application_admin_repo.get_application_admin_by_app_and_user_id( + request.application_id, fam_user.user_id + ) + ) + if fam_application_admin_user: + LOGGER.debug( + "FamApplicationAdmin already exists with id: " + + f"{fam_application_admin_user.application_admin_id}." + ) + error_msg = "User is admin already." + utils.raise_http_exception(HTTPStatus.CONFLICT, error_msg) + else: + # Create application admin if user is not admin yet + fam_application_admin_user = ( + self.application_admin_repo.create_application_admin( + request.application_id, fam_user.user_id, requester + ) + ) + + fam_application_admin_user_dict = fam_application_admin_user.__dict__ + app_admin_user_assignment = schemas.FamAppAdminGet( + **fam_application_admin_user_dict + ) + LOGGER.debug( + f"Application admin user assignment executed successfully: {app_admin_user_assignment}" + ) + return app_admin_user_assignment + + def delete_application_admin(self, application_admin_id: int): + return self.application_admin_repo.delete_application_admin( + application_admin_id + ) diff --git a/server/admin_management/api/app/services/application_service.py b/server/admin_management/api/app/services/application_service.py new file mode 100644 index 000000000..aa28657a2 --- /dev/null +++ b/server/admin_management/api/app/services/application_service.py @@ -0,0 +1,15 @@ +import logging +from sqlalchemy.orm import Session + +from api.app.repositories.application_repository import ApplicationRepository + + +LOGGER = logging.getLogger(__name__) + + +class ApplicationService: + def __init__(self, db: Session): + self.application_repo = ApplicationRepository(db) + + def get_application(self, application_id: int): + return self.application_repo.get_application(application_id) \ No newline at end of file diff --git a/server/admin_management/api/app/services/user_service.py b/server/admin_management/api/app/services/user_service.py new file mode 100644 index 000000000..c48c0cea7 --- /dev/null +++ b/server/admin_management/api/app/services/user_service.py @@ -0,0 +1,41 @@ +import logging +from sqlalchemy.orm import Session + +from api.app import schemas +from api.app.repositories.user_repository import UserRepository + + +LOGGER = logging.getLogger(__name__) + + +class UserService: + def __init__(self, db: Session): + self.user_repo = UserRepository(db) + + def get_user_by_domain_and_name(self, user_type_code: str, user_name: str): + return self.user_repo.get_user_by_domain_and_name(user_type_code, user_name) + + def get_user_by_cognito_user_id(self, cognito_user_id: str): + return self.user_repo.get_user_by_cognito_user_id(cognito_user_id) + + def find_or_create(self, user_type_code: str, user_name: str, requester: str): + LOGGER.debug( + f"Request for finding or creating a user with user_type: {user_type_code}, " + + f"user_name: {user_name}." + ) + + fam_user = self.get_user_by_domain_and_name(user_type_code, user_name) + if not fam_user: + request_user = schemas.FamUser( + **{ + "user_type_code": user_type_code, + "user_name": user_name, + "create_user": requester, + } + ) + fam_user = self.user_repo.create_user(request_user) + LOGGER.debug(f"User created: {fam_user.user_id}.") + return fam_user + + LOGGER.debug(f"User {fam_user.user_id} found.") + return fam_user diff --git a/server/admin_management/api/app/utils/__init__.py b/server/admin_management/api/app/utils/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/server/admin_management/api/app/utils/audit_util.py b/server/admin_management/api/app/utils/audit_util.py new file mode 100644 index 000000000..18f890064 --- /dev/null +++ b/server/admin_management/api/app/utils/audit_util.py @@ -0,0 +1,98 @@ +import logging +from enum import Enum +import json +from api.app.models import model as models +from fastapi import Request, HTTPException + + +LOGGER = logging.getLogger(__name__) + + +class AuditEventType(str, Enum): + CREATE_APPLICATION_ADMIN_ACCESS = "Grant User Application Admin Access" + REMOVE_APPLICATION_ADMIN_ACCESS = "Remove User Application Admin Access" + + +class AuditEventOutcome(str, Enum): + SUCCESS = 1 + FAIL = 0 + + +class AuditEventLog: + request: Request + event_type: AuditEventType + event_outcome: AuditEventOutcome + application: models.FamApplication + requesting_user: models.FamUser + target_user: models.FamUser + exception: Exception + + def __init__( + self, + request: Request = None, + event_type: AuditEventType = None, + event_outcome: AuditEventOutcome = None, + application: models.FamApplication = None, + requesting_user: models.FamUser = None, + target_user: models.FamUser = None, + exception: Exception = None, + ): + self.request = request + self.event_type = event_type + self.event_outcome = event_outcome + self.application = application + self.requesting_user = requesting_user + self.target_user = target_user + self.exception = exception + + def log_event(self): + + log_item = { + "auditEventTypeCode": self.event_type.name if self.event_type else None, + "auditEventResultCode": self.event_outcome.name + if self.event_outcome + else None, + "applicationId": self.application.application_id + if self.application + else None, + "applicationName": self.application.application_name + if self.application + else None, + "applicationEnv": self.application.app_environment + if self.application + else None, + "targetUser": { + "userGuid": self.target_user.user_guid if self.target_user else None, + "userType": self.target_user.user_type_code + if self.target_user + else None, + "idpUserName": self.target_user.user_name if self.target_user else None, + "cognitoUsername": self.target_user.cognito_user_id + if self.target_user + else None, + }, + "requestingUser": { + "userGuid": self.requesting_user.user_guid + if self.requesting_user + else None, + "userType": self.requesting_user.user_type_code + if self.requesting_user + else None, + "idpUserName": self.requesting_user.user_name + if self.requesting_user + else None, + "cognitoUsername": self.requesting_user.cognito_user_id + if self.requesting_user + else None, + }, + "requestIP": self.request.client.host if self.request.client else "unknown", + } + + if self.exception and type(self.exception) == HTTPException: + log_item["exception"] = { + "exceptionType": "HTTPException", + "statusCode": self.exception.status_code, + "details": self.exception.detail, + } + + LOGGER.info(json.dumps(log_item)) diff --git a/server/admin_management/api/app/utils/utils.py b/server/admin_management/api/app/utils/utils.py new file mode 100644 index 000000000..483458e19 --- /dev/null +++ b/server/admin_management/api/app/utils/utils.py @@ -0,0 +1,9 @@ +import logging +from fastapi import HTTPException + +LOGGER = logging.getLogger(__name__) + + +def raise_http_exception(status_code: str, error_msg: str): + LOGGER.info(error_msg) + raise HTTPException(status_code=status_code, detail=error_msg) diff --git a/server/admin_management/requirements.txt b/server/admin_management/requirements.txt index 4b7844ed7..51d86f188 100644 --- a/server/admin_management/requirements.txt +++ b/server/admin_management/requirements.txt @@ -5,4 +5,6 @@ mangum==0.17.0 boto3==1.28.83 psycopg2-binary==2.9.7 python-jose==3.3.0 -httpx==0.24.1 \ No newline at end of file +httpx==0.24.1 +pydantic==2.4.2 +pydantic_core==2.10.1 \ No newline at end of file diff --git a/server/flyway/sql/V33__add_permission_for_admin_management_api_user.sql b/server/flyway/sql/V33__add_permission_for_admin_management_api_user.sql new file mode 100644 index 000000000..2a0c055fd --- /dev/null +++ b/server/flyway/sql/V33__add_permission_for_admin_management_api_user.sql @@ -0,0 +1,7 @@ +-- -- on 'fam_application_admin' table +GRANT SELECT, UPDATE, DELETE, INSERT ON app_fam.fam_user_type_code TO ${admin_management_api_db_user} +; + +ALTER TABLE app_fam.fam_application_admin + ALTER COLUMN create_user SET DATA TYPE VARCHAR(60), + ALTER COLUMN update_user SET DATA TYPE VARCHAR(60);