fix: adjust permission check logic to avoid sso_verified apps

api/controllers/web/app.py

@@ -1,15 +1,16 @@
-from flask import request
-from flask_restful import Resource, marshal_with, reqparse
-
 from controllers.common import fields
 from controllers.web import api
 from controllers.web.error import AppUnavailableError
 from controllers.web.wraps import WebApiResource
-from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
+from core.app.app_config.common.parameters_mapping import \
+    get_parameters_from_feature_dict
+from flask import request
+from flask_restful import Resource, marshal_with, reqparse
 from libs.passport import PassportService
 from models.model import App, AppMode
 from services.app_service import AppService
 from services.enterprise.enterprise_service import EnterpriseService
+from services.webapp_auth_service import WebAppAuthService
 
 
 class AppParameterApi(WebApiResource):
@@ -90,7 +91,9 @@ class AppWebAuthPermission(Resource):
         app_id = args["appId"]
         app_code = AppService.get_app_code_by_id(app_id)
 
-        res = EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(str(user_id), app_code)
+        res = True
+        if WebAppAuthService.is_app_require_permission_check(app_id=app_id):
+            res = EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(str(user_id), app_code)
         return {"result": res}
 
 

api/controllers/web/login.py

@@ -1,12 +1,11 @@
+from flask_restful import Resource, reqparse
+from jwt import InvalidTokenError  # type: ignore
+
 import services
-from controllers.console.auth.error import (EmailCodeError,
-                                            EmailOrPasswordMismatchError,
-                                            InvalidEmailError)
+from controllers.console.auth.error import EmailCodeError, EmailOrPasswordMismatchError, InvalidEmailError
 from controllers.console.error import AccountBannedError, AccountNotFound
 from controllers.console.wraps import only_edition_enterprise, setup_required
 from controllers.web import api
-from flask_restful import Resource, reqparse
-from jwt import InvalidTokenError  # type: ignore
 from libs.helper import email
 from libs.password import valid_password
 from services.account_service import AccountService

api/controllers/web/passport.py

@@ -21,13 +21,13 @@ class PassportResource(Resource):
         system_features = FeatureService.get_system_features()
         app_code = request.headers.get("X-App-Code")
         user_id = request.args.get("user_id")
-        enterprise_login_token = request.args.get("enterprise_login_token")
+        web_app_access_token = request.args.get("web_app_access_token")
 
         if app_code is None:
             raise Unauthorized("X-App-Code header is missing.")
 
         # exchange token for enterprise logined web user
-        enterprise_user_decoded = decode_enterprise_webapp_user_id(enterprise_login_token)
+        enterprise_user_decoded = decode_enterprise_webapp_user_id(web_app_access_token)
         if enterprise_user_decoded:
             # a web user has already logged in, exchange a token for this app without redirecting to the login page
             return exchange_token_for_existing_web_user(
@@ -122,7 +122,9 @@ def exchange_token_for_existing_web_user(app_code: str, enterprise_user_decoded:
     app_model = db.session.query(App).filter(App.id == site.app_id).first()
     if not app_model or app_model.status != "normal" or not app_model.enable_site:
         raise NotFound()
-    end_user = db.session.query(EndUser).filter(EndUser.id == end_user_id).first()
+    end_user = None
+    if end_user_id:
+        end_user = db.session.query(EndUser).filter(EndUser.id == end_user_id).first()
     if not end_user:
         end_user = EndUser(
             tenant_id=app_model.tenant_id,

api/controllers/web/wraps.py

@@ -1,15 +1,17 @@
 from functools import wraps
 
+from controllers.web.error import (WebAppAuthAccessDeniedError,
+                                   WebAppAuthRequiredError)
+from extensions.ext_database import db
 from flask import request
 from flask_restful import Resource
-from werkzeug.exceptions import BadRequest, NotFound, Unauthorized
-
-from controllers.web.error import WebAppAuthAccessDeniedError, WebAppAuthRequiredError
-from extensions.ext_database import db
 from libs.passport import PassportService
 from models.model import App, EndUser, Site
-from services.enterprise.enterprise_service import EnterpriseService, WebAppSettings
+from services.enterprise.enterprise_service import (EnterpriseService,
+                                                    WebAppSettings)
 from services.feature_service import FeatureService
+from services.webapp_auth_service import WebAppAuthService
+from werkzeug.exceptions import BadRequest, NotFound, Unauthorized
 
 
 def validate_jwt_token(view=None):
@@ -45,7 +47,8 @@ def decode_jwt_token():
             raise Unauthorized("Invalid Authorization header format. Expected 'Bearer <api-key>' format.")
         decoded = PassportService().verify(tk)
         app_code = decoded.get("app_code")
-        app_model = db.session.query(App).filter(App.id == decoded["app_id"]).first()
+        app_id = decoded.get("app_id")
+        app_model = db.session.query(App).filter(App.id == app_id).first()
         site = db.session.query(Site).filter(Site.code == app_code).first()
         if not app_model:
             raise NotFound()
@@ -53,7 +56,8 @@ def decode_jwt_token():
             raise BadRequest("Site URL is no longer valid.")
         if app_model.enable_site is False:
             raise BadRequest("Site is disabled.")
-        end_user = db.session.query(EndUser).filter(EndUser.id == decoded["end_user_id"]).first()
+        end_user_id = decoded.get("end_user_id")
+        end_user = db.session.query(EndUser).filter(EndUser.id == end_user_id).first()
         if not end_user:
             raise NotFound()
 
@@ -115,9 +119,7 @@ def _validate_user_accessibility(
         if not webapp_settings:
             raise WebAppAuthRequiredError("Web app settings not found.")
 
-        access_modes_require_permission_check = ["private", "private_all"]
-
-        if webapp_settings.access_mode in access_modes_require_permission_check:
+        if WebAppAuthService.is_app_require_permission_check(access_mode=webapp_settings.access_mode):
             if not EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(user_id, app_code=app_code):
                 raise WebAppAuthAccessDeniedError()
 

api/services/webapp_auth_service.py

@@ -9,6 +9,8 @@ from libs.passport import PassportService
 from libs.password import compare_password
 from models.account import Account, AccountStatus
 from models.model import App, EndUser, Site
+from services.app_service import AppService
+from services.enterprise.enterprise_service import EnterpriseService
 from services.errors.account import (AccountLoginError, AccountNotFoundError,
                                      AccountPasswordError)
 from tasks.mail_email_code_login import send_email_code_login_mail_task
@@ -60,7 +62,7 @@ class WebAppAuthService:
 
         code = "".join([str(random.randint(0, 9)) for _ in range(6)])
         token = TokenManager.generate_token(
-            account=account, email=email, token_type="webapp_email_code_login", additional_data={"code": code}
+            account=account, email=email, token_type="email_code_login", additional_data={"code": code}
         )
         send_email_code_login_mail_task.delay(
             language=language,
@@ -72,11 +74,11 @@ class WebAppAuthService:
 
     @classmethod
     def get_email_code_login_data(cls, token: str) -> Optional[dict[str, Any]]:
-        return TokenManager.get_token_data(token, "webapp_email_code_login")
+        return TokenManager.get_token_data(token, "email_code_login")
 
     @classmethod
     def revoke_email_code_login_token(cls, token: str):
-        TokenManager.revoke_token(token, "webapp_email_code_login")
+        TokenManager.revoke_token(token, "email_code_login")
 
     @classmethod
     def create_end_user(cls, app_code, email) -> EndUser:
@@ -114,3 +116,28 @@ class WebAppAuthService:
 
         token: str = PassportService().issue(payload)
         return token
+
+    @classmethod
+    def is_app_require_permission_check(cls, app_code: str = None, app_id: str = None, access_mode: str = None) -> bool:
+        """
+        Check if the app requires permission check based on its access mode.
+        """
+        modes_requiring_permission_check = [
+            "private",
+            "private_all",
+        ]
+        if access_mode:
+            return access_mode in modes_requiring_permission_check
+
+        if not app_code and not app_id:
+            raise ValueError("Either app_code or app_id must be provided.")
+
+        if app_code:
+            app_id = AppService.get_app_id_by_code(app_code)
+        if not app_id:
+            raise ValueError("App ID could not be determined from the provided app_code.")
+
+        webapp_settings = EnterpriseService.WebAppAuth.get_app_access_mode_by_id(app_id)
+        if webapp_settings and webapp_settings.access_mode in modes_requiring_permission_check:
+            return True
+        return False