fix: use available email verification claims (#528)

safety/auth/cli.py

@@ -3,6 +3,7 @@
 import sys
 from safety.auth.models import Auth
 
+from safety.auth.utils import is_email_verified
 from safety.console import main_console as console
 from safety.constants import MSG_FINISH_REGISTRATION_TPL, MSG_VERIFICATION_HINT
 
@@ -207,7 +208,7 @@ def status(ctx: typer.Context, ensure_auth: bool = False,
         console.print()
 
     if info:
-        verified = info.get("email_verified", False)
+        verified = is_email_verified(info)
         email_status = " [red](email not verified)[/red]" if not verified else ""
 
         console.print(f'[green]Authenticated as {info["email"]}[/green]{email_status}')    

safety/auth/cli_utils.py

@@ -9,7 +9,7 @@
 from safety.auth.constants import CLIENT_ID, OPENID_CONFIG_URL
 
 from safety.auth.models import Organization, Auth
-from safety.auth.utils import S3PresignedAdapter, SafetyAuthSession, get_keys
+from safety.auth.utils import S3PresignedAdapter, SafetyAuthSession, get_keys, is_email_verified
 from safety.constants import REQUEST_TIMEOUT
 from safety.scan.constants import CLI_KEY_HELP, CLI_PROXY_HOST_HELP, CLI_PROXY_PORT_HELP, CLI_PROXY_PROTOCOL_HELP, CLI_STAGE_HELP
 from safety.scan.util import Stage
@@ -176,7 +176,7 @@ def inner(ctx, proxy_protocol: Optional[str] = None,
         if info:
             ctx.obj.auth.name = info.get("name")
             ctx.obj.auth.email = info.get("email")
-            ctx.obj.auth.email_verified = info.get("email_verified", False)
+            ctx.obj.auth.email_verified = is_email_verified(info)
             SafetyContext().account = info["email"]
         else:
             SafetyContext().account = ""

safety/auth/constants.py

@@ -14,6 +14,9 @@
 
 OPENID_CONFIG_URL = f"{AUTH_SERVER_URL}/.well-known/openid-configuration"
 
+CLAIM_EMAIL_VERIFIED_API = "https://api.safetycli.com/email_verified"
+CLAIM_EMAIL_VERIFIED_AUTH_SERVER = "email_verified"
+
 CLI_AUTH = f'{SAFETY_PLATFORM_URL}/cli/auth'
 CLI_AUTH_SUCCESS = f'{SAFETY_PLATFORM_URL}/cli/auth/success'
 CLI_AUTH_LOGOUT = f'{SAFETY_PLATFORM_URL}/cli/logout'

safety/auth/main.py

@@ -54,15 +54,17 @@ def get_organization() -> Optional[Organization]:
     return org
 
 def get_auth_info(ctx):
+    from safety.auth.utils import is_email_verified
+
     info = None
     if ctx.obj.auth.client.token:
         try:
             info = get_token_data(get_token(name='id_token'), keys=ctx.obj.auth.keys)
 
-            verified = info.get("email_verified", False)
+            verified = is_email_verified(info)
             if not verified:
                 user_info = ctx.obj.auth.client.fetch_user_info()
-                verified = user_info.get("email_verified", False)
+                verified = is_email_verified(user_info)
 
                 if verified:
                     # refresh only if needed 

safety/auth/models.py

@@ -39,9 +39,11 @@ def is_valid(self) -> bool:
         return bool(self.client.token and self.email_verified)
 
     def refresh_from(self, info):
+        from safety.auth.utils import is_email_verified
+
         self.name = info.get("name")
         self.email = info.get("email")
-        self.email_verified = info.get("email_verified", False)
+        self.email_verified = is_email_verified(info)
 
 class XAPIKeyAuth(BaseOAuth):
     def __init__(self, api_key):

safety/auth/utils.py

@@ -5,7 +5,8 @@
 from authlib.integrations.base_client.errors import OAuthError
 import requests
 from requests.adapters import HTTPAdapter
-from safety.auth.constants import AUTH_SERVER_URL
+from safety.auth.constants import AUTH_SERVER_URL, CLAIM_EMAIL_VERIFIED_API, \
+    CLAIM_EMAIL_VERIFIED_AUTH_SERVER
 from safety.auth.main import get_auth_info, get_token_data
 from safety.constants import PLATFORM_API_CHECK_UPDATES_ENDPOINT, PLATFORM_API_INITIALIZE_SCAN_ENDPOINT, PLATFORM_API_POLICY_ENDPOINT, \
     PLATFORM_API_PROJECT_CHECK_ENDPOINT, PLATFORM_API_PROJECT_ENDPOINT, PLATFORM_API_PROJECT_SCAN_REQUEST_ENDPOINT, \
@@ -19,11 +20,17 @@
 
 LOG = logging.getLogger(__name__)
 
+
 def get_keys(client_session, openid_config):
     if "jwks_uri" in openid_config:
         return client_session.get(url=openid_config["jwks_uri"], bearer=False).json()
     return None
 
+
+def is_email_verified(info) -> bool:
+    return info.get(CLAIM_EMAIL_VERIFIED_API) or info.get(CLAIM_EMAIL_VERIFIED_AUTH_SERVER)
+
+
 def parse_response(func):
     def wrapper(*args, **kwargs):
         try: