"""Logic for JWTs.""" from typing import Any, Dict, Optional from fastapi import HTTPException from pdp import config from pdp.constants import error from pdp.constants.constants import USER_TYPE_HUMAN, USER_TYPE_MACHINE CLAIM_ORCHARD_IDENTITY_ID = "orchardIdentityId" AUTHORIZATION = "Authorization" def get_identity_uuid(decoded_jwt: Dict[str, Any]) -> Optional[str]: """Get a user's orchardIdentityId from a decoded JWT.""" identity_uuid = None grass_data = decoded_jwt.get(config.JWT_USER_METADATA) if grass_data: identity_uuid = grass_data.get(CLAIM_ORCHARD_IDENTITY_ID) return identity_uuid def get_user_type(decoded_jwt: Dict[str, Any]) -> str: """Get a user's userType from a decoded JWT.""" grass_data = decoded_jwt.get(config.JWT_USER_METADATA) if grass_data: if grass_data.get(config.IS_MACHINE) is True: return USER_TYPE_MACHINE return USER_TYPE_HUMAN def get_impersonated_by_identity_uuid(decoded_jwt: Dict[str, Any]) -> Optional[str]: """Get the impersonator's orchardIdentityId from a decoded JWT.""" if config.JWT_IMPERSONATED_BY not in decoded_jwt: return None impersonated_by_identity_uuid: str | None = decoded_jwt.get( config.JWT_IMPERSONATED_BY ) if not impersonated_by_identity_uuid: raise HTTPException( status_code=401, detail=error.ERROR_MESSAGE_BAD_IMPERSONATED_BY_IDENTITY_UUID, ) return impersonated_by_identity_uuid