o ?7if)@sXUdZddlmZddlmZmZddlmZmZddlm Z ddl Z ddl Z ddl m Z edgd d Zdae e jed <d e jfd dZdeded efddZded efddZdRdede ed efddZded e efddZeZd efddZded eeeffdd Zded!ed dfd"d#Zded efd$d%Z d efd&d'Z!d efd(d)Z"d efd*d+Z#d efd,d-Z$d efd.d/Z%d efd0d1Z&d efd2d3Z'd efd4d5Z(d efd6d7Z)d efd8d9Z*d:e ed efd;d<Z+dRd=ed>e ed efd?d@Z,dRdAe ed efdBdCZ-dDdEZ.dRdSdHdIZ/d e0fdJdKZ1dLe0d efdMdNZ2dOed efdPdQZ3dS)TzM Security utilities JWT token handling, password hashing, and authentication ) CryptContext)JWTErrorjwt)datetime timedelta)OptionalN)settingsbcryptauto)schemes deprecated _redis_clientreturncCsBtdurtjrtjtjddatStjtjtjtjtj ddatS)z/Get or create Redis client for token managementNT)decode_responses)hostportdbpasswordr) r r REDIS_URLredisfrom_urlRedis REDIS_HOST REDIS_PORTREDIS_DBREDIS_PASSWORDrr9/lsinfo/ai/hellotax_ai/base_platform/app/core/security.pyget_redis_clients rplain_passwordhashed_passwordcCs2t|tr|dddjddd}t||S)z* Verify a password against a hash zutf-8NHignore)errors) isinstancestrencodedecode pwd_contextverify)rr rrrverify_passwords  r*rcCs t|S)z& Hash a password using bcrypt )r(hashrrrrget_password_hash)s r-data expires_deltacCsT|}|r t|}n tttjd}|d|itj|tj tj d}|S)z Create a JWT access token Args: data: Dictionary containing token payload expires_delta: Optional expiration time delta Returns: Encoded JWT token string )minutesexp) algorithm) copyrutcnowrrACCESS_TOKEN_EXPIRE_MINUTESupdaterr& SECRET_KEY ALGORITHM)r.r/ to_encodeexpire encoded_jwtrrrcreate_access_token0s r<tokencCs2ztj|tjtjgd}|WStyYdSw)z Decode and verify a JWT access token Args: token: JWT token string Returns: Token payload dict if valid, None otherwise ) algorithmsN)rr'rr7r8r)r=payloadrrrdecode_access_tokenFs   r@cCs tdS)zx Generate a random verification token for email verification Returns: URL-safe random token string )secrets token_urlsaferrrrgenerate_verification_token[s rDcstt|dkrdStdd|DsdStdd|DsdStdd|Ds)d Sd tfd d|Ds8d Sd S)a Validate password strength Requirements: - At least 8 characters - Contains uppercase letter - Contains lowercase letter - Contains digit - Contains special character Returns: Tuple of (is_valid, error_message) )Fu密码至少需要8个字符cs|]}|VqdSN)isupper.0crrr vz-validate_password_strength..)Fu密码必须包含大写字母csrFrG)islowerrIrrrrLyrM)Fu密码必须包含小写字母csrFrG)isdigitrIrrrrL|rM)Fu密码必须包含数字z!@#$%^&*()_+-=[]{}|;:,.<>?c3|]}|vVqdSrGrrI special_charsrrrLrM)Fu密码必须包含特殊字符)T)lenanyr,rrQrvalidate_password_strengthes rV exp_timestampcCs>t}|tt}|dkr|d||ddSdS)z Revoke a JWT token by adding it to Redis blacklist Args: token: JWT token string exp_timestamp: Token expiration timestamp rrevoked_token:1N)rintrr4 timestampsetex)r=rWclientttlrrr revoke_tokens r_c Csfzt}|d|dkWSty2}zddl}|t}|d|WYd}~dSd}~ww)z Check if a token has been revoked Args: token: JWT token string Returns: True if token is revoked rXrNz.Redis unavailable for token revocation check: T)rexists Exceptionlogging getLogger__name__error)r=r]erbloggerrrris_token_revokeds  rhcCsFt|dr|jr|jD]}|js|jjs|jjSq |jr!|jSdS)z Get user's primary role (backward compatible) Priority: 1. user_roles relationship (first non-deleted role) 2. Fallback to user.role field Args: user: User object Returns: Role code string user_roles customer_user)hasattrri is_deletedrolecode)userurrrrget_primary_roles   rqcC t|dkS)z'Check if user is platform administratorplatform_adminrqrorrris_platform_admin rvcCrr)z*Check if user is platform user (read-only) platform_userrtrurrris_platform_userrwrycCrr)z'Check if user is customer administratorcustomer_adminrtrurrris_customer_adminrwr{cCrr)z&Check if user is customer regular userrjrtrurrris_customer_userrwr|cC t|dvS)z'Check if user has platform-level accessrsrxrtrurrrhas_platform_accessrwrcCr})z5Check if user has admin access (platform or customer)rsrzrtrurrrhas_admin_accessrwrcCr})z$Check if user can manage other usersrrtrurrrcan_manage_usersrwrcCr})z"Check if user can delete resourcesrrtrurrrcan_delete_resourcerwrcCr})z$Check if user can view all resourcesr~rtrurrrcan_view_all_resourcesrwrtarget_tenant_idcCs$t|rdS|jdur dS|j|kS)z Check if user can access resources from a specific company Args: user: User object target_tenant_id: Target company ID to check access for Returns: True if user can access the company's resources TNF)r tenant_id)rorrrrcan_access_companys    rresource_owner_idresource_tenant_idcCs4t|rdS||jkr dSt|r|r|j|kSdS)a# Check if user can manage a specific resource Args: user: User object resource_owner_id: ID of the user who owns the resource resource_tenant_id: Optional company ID associated with the resource Returns: True if user can manage the resource TF)rvidr{r)rorrrrrcan_manage_resources    r company_namecCs>ddddd}||j|j}|r|jdvr|d|S|S)z Get display string for user role Args: user: User object company_name: Optional company name Returns: Formatted role display string u平台管理员u 平台用户u企业管理员u 企业用户)rsrxrzrj)rzrj/)getrm)ror role_names role_displayrrrget_role_displays rcCs*ddlm}m}t|s||jdddS)zW Raise exception if user is not an admin Used as a simple permission check r HTTPExceptionstatuszAdmin access required status_codedetailN)fastapirrrHTTP_403_FORBIDDEN)rorrrrr require_admin6sr current_userUsercs^ddlmmm}ddlm}|dur"||fd fdd }|St|s-jd d |S) z5 Dependency to require platform admin access r)rrDepends)get_current_userNrorcst|s jdd|S)NPlatform admin access requiredr)rvrrurrr_require_platform_adminMs z7require_platform_admin.._require_platform_adminrr)ror)rrrr app.api.depsrrvr)rrrrrrrrequire_platform_adminDs rcCst|dr dd|jDSgS)z>Get list of role codes for a user from user_roles relationshipricSs"g|] }|js|jjs|jjqSr)rlrmrn)rJrprrr cs"z"get_user_roles..)rkrirurrrget_user_roles`s r role_codescs,t|}|s |jvStfdd|DS)z,Check if user has any of the specified rolesc3rPrGr)rJrnrrrrLmrMzhas_any_role..)rrmrU)roruser_role_codesrrr has_any_rolegs r role_codecCs t||gS)z!Check if user has a specific role)r)rorrrrhas_roleprwrrG)rr)4__doc__passlib.contextrjoserrrrtypingrrBr app.configrr(r r__annotations__rr%boolr*r-dictr<r@ verify_tokenrDtuplerVrZr_rhrqrvryr{r|rrrrrrrrrrlistrrrrrrrsL      !