fix: use LicenseStatus enum instead of raw strings and tighten path prefix matching

Replace raw license status strings with LicenseStatus enum values in
app_factory.py and enterprise_service.py to prevent silent mismatches.
Use trailing-slash prefixes ('/console/api/', '/api/') to avoid false
matches on unrelated paths like /api-docs.

api/app_factory.py

@@ -1,13 +1,17 @@
 import logging
 import time
 
+from flask import request
 from opentelemetry.trace import get_current_span
 from opentelemetry.trace.span import INVALID_SPAN_ID, INVALID_TRACE_ID
 
 from configs import dify_config
 from contexts.wrapper import RecyclableContextVar
+from controllers.console.error import UnauthorizedAndForceLogout
 from core.logging.context import init_request_context
 from dify_app import DifyApp
+from services.enterprise.enterprise_service import EnterpriseService
+from services.feature_service import LicenseStatus
 
 logger = logging.getLogger(__name__)
 
@@ -31,6 +35,38 @@ def create_flask_app_with_configs() -> DifyApp:
         init_request_context()
         RecyclableContextVar.increment_thread_recycles()
 
+        # Enterprise license validation for API endpoints (both console and webapp)
+        # When license expires, block all API access except bootstrap endpoints needed
+        # for the frontend to load the license expiration page without infinite reloads.
+        if dify_config.ENTERPRISE_ENABLED:
+            is_console_api = request.path.startswith("/console/api/")
+            is_webapp_api = request.path.startswith("/api/") and not is_console_api
+
+            if is_console_api or is_webapp_api:
+                if is_console_api:
+                    console_exempt_prefixes = (
+                        "/console/api/system-features",
+                        "/console/api/setup",
+                        "/console/api/version",
+                        "/console/api/activate/check",
+                    )
+                    is_exempt = any(request.path.startswith(p) for p in console_exempt_prefixes)
+                else:  # webapp API
+                    is_exempt = request.path.startswith("/api/system-features")
+
+                if not is_exempt:
+                    try:
+                        # Check license status with caching (10 min TTL)
+                        license_status = EnterpriseService.get_cached_license_status()
+                        if license_status in (LicenseStatus.INACTIVE, LicenseStatus.EXPIRED, LicenseStatus.LOST):
+                            raise UnauthorizedAndForceLogout(
+                                f"Enterprise license is {license_status}. Please contact your administrator."
+                            )
+                    except UnauthorizedAndForceLogout:
+                        raise
+                    except Exception:
+                        logger.exception("Failed to check enterprise license status")
+
     # add after request hook for injecting trace headers from OpenTelemetry span context
     # Only adds headers when OTEL is enabled and has valid context
     @dify_app.after_request

api/services/enterprise/base.py

@@ -6,6 +6,13 @@ from typing import Any
 import httpx
 
 from core.helper.trace_id_helper import generate_traceparent_header
+from services.errors.enterprise import (
+    EnterpriseAPIBadRequestError,
+    EnterpriseAPIError,
+    EnterpriseAPIForbiddenError,
+    EnterpriseAPINotFoundError,
+    EnterpriseAPIUnauthorizedError,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -64,10 +71,56 @@ class BaseRequest:
                 request_kwargs["timeout"] = timeout
 
             response = client.request(method, url, **request_kwargs)
+
+            # Always validate HTTP status and raise domain-specific errors
+            if not response.is_success:
+                cls._handle_error_response(response)
+
+            # Legacy support: still respect raise_for_status parameter
             if raise_for_status:
                 response.raise_for_status()
+
         return response.json()
 
+    @classmethod
+    def _handle_error_response(cls, response: httpx.Response) -> None:
+        """
+        Handle non-2xx HTTP responses by raising appropriate domain errors.
+
+        Attempts to extract error message from JSON response body,
+        falls back to status text if parsing fails.
+        """
+        error_message = f"Enterprise API request failed: {response.status_code} {response.reason_phrase}"
+
+        # Try to extract error message from JSON response
+        try:
+            error_data = response.json()
+            if isinstance(error_data, dict):
+                # Common error response formats:
+                # {"error": "...", "message": "..."}
+                # {"message": "..."}
+                # {"detail": "..."}
+                error_message = (
+                    error_data.get("message") or error_data.get("error") or error_data.get("detail") or error_message
+                )
+        except Exception:
+            # If JSON parsing fails, use the default message
+            logger.debug(
+                "Failed to parse error response from enterprise API (status=%s)", response.status_code, exc_info=True
+            )
+
+        # Raise specific error based on status code
+        if response.status_code == 400:
+            raise EnterpriseAPIBadRequestError(error_message)
+        elif response.status_code == 401:
+            raise EnterpriseAPIUnauthorizedError(error_message)
+        elif response.status_code == 403:
+            raise EnterpriseAPIForbiddenError(error_message)
+        elif response.status_code == 404:
+            raise EnterpriseAPINotFoundError(error_message)
+        else:
+            raise EnterpriseAPIError(error_message, status_code=response.status_code)
+
 
 class EnterpriseRequest(BaseRequest):
     base_url = os.environ.get("ENTERPRISE_API_URL", "ENTERPRISE_API_URL")

api/services/enterprise/enterprise_service.py

@@ -5,11 +5,15 @@ from datetime import datetime
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from configs import dify_config
+from extensions.ext_redis import redis_client
 from services.enterprise.base import EnterpriseRequest
 
 logger = logging.getLogger(__name__)
 
 DEFAULT_WORKSPACE_JOIN_TIMEOUT_SECONDS = 1.0
+# License status cache configuration
+LICENSE_STATUS_CACHE_KEY = "enterprise:license:status"
+LICENSE_STATUS_CACHE_TTL = 600  # 10 minutes
 
 
 class WebAppSettings(BaseModel):
@@ -223,3 +227,47 @@ class EnterpriseService:
 
             params = {"appId": app_id}
             EnterpriseRequest.send_request("DELETE", "/webapp/clean", params=params)
+
+    @classmethod
+    def get_cached_license_status(cls):
+        """
+        Get enterprise license status with Redis caching to reduce HTTP calls.
+
+        Only caches valid statuses (active/expiring) since invalid statuses
+        should be re-checked every request — the admin may update the license
+        at any time.
+
+        Returns license status string or None if unavailable.
+        """
+        if not dify_config.ENTERPRISE_ENABLED:
+            return None
+
+        # Try cache first — only valid statuses are cached
+        try:
+            cached_status = redis_client.get(LICENSE_STATUS_CACHE_KEY)
+            if cached_status:
+                if isinstance(cached_status, bytes):
+                    cached_status = cached_status.decode("utf-8")
+                return cached_status
+        except Exception:
+            logger.debug("Failed to get license status from cache, calling enterprise API")
+
+        # Cache miss or failure — call enterprise API
+        try:
+            info = cls.get_info()
+            license_info = info.get("License")
+            if license_info:
+                from services.feature_service import LicenseStatus
+
+                status = license_info.get("status", LicenseStatus.INACTIVE)
+                # Only cache valid statuses so license updates are picked up immediately
+                if status in (LicenseStatus.ACTIVE, LicenseStatus.EXPIRING):
+                    try:
+                        redis_client.setex(LICENSE_STATUS_CACHE_KEY, LICENSE_STATUS_CACHE_TTL, status)
+                    except Exception:
+                        logger.debug("Failed to cache license status")
+                return status
+        except Exception:
+            logger.exception("Failed to get enterprise license status")
+
+        return None

api/services/errors/__init__.py

@@ -7,6 +7,7 @@ from . import (
     conversation,
     dataset,
     document,
+    enterprise,
     file,
     index,
     message,
@@ -21,6 +22,7 @@ __all__ = [
     "conversation",
     "dataset",
     "document",
+    "enterprise",
     "file",
     "index",
     "message",

api/services/errors/enterprise.py

@@ -0,0 +1,45 @@
+"""Enterprise service errors."""
+
+from services.errors.base import BaseServiceError
+
+
+class EnterpriseServiceError(BaseServiceError):
+    """Base exception for enterprise service errors."""
+
+    def __init__(self, description: str | None = None, status_code: int | None = None):
+        super().__init__(description)
+        self.status_code = status_code
+
+
+class EnterpriseAPIError(EnterpriseServiceError):
+    """Generic enterprise API error (non-2xx response)."""
+
+    pass
+
+
+class EnterpriseAPINotFoundError(EnterpriseServiceError):
+    """Enterprise API returned 404 Not Found."""
+
+    def __init__(self, description: str | None = None):
+        super().__init__(description, status_code=404)
+
+
+class EnterpriseAPIForbiddenError(EnterpriseServiceError):
+    """Enterprise API returned 403 Forbidden."""
+
+    def __init__(self, description: str | None = None):
+        super().__init__(description, status_code=403)
+
+
+class EnterpriseAPIUnauthorizedError(EnterpriseServiceError):
+    """Enterprise API returned 401 Unauthorized."""
+
+    def __init__(self, description: str | None = None):
+        super().__init__(description, status_code=401)
+
+
+class EnterpriseAPIBadRequestError(EnterpriseServiceError):
+    """Enterprise API returned 400 Bad Request."""
+
+    def __init__(self, description: str | None = None):
+        super().__init__(description, status_code=400)

api/services/feature_service.py

@@ -379,11 +379,14 @@ class FeatureService:
             )
             features.webapp_auth.sso_config.protocol = enterprise_info.get("SSOEnforcedForWebProtocol", "")
 
-        if is_authenticated and (license_info := enterprise_info.get("License")):
+        # License status and expiry are always exposed so the login page can
+        # show the expiry UI after a force-logout (the user is unauthenticated
+        # at that point).  Workspace usage details remain auth-gated.
+        if license_info := enterprise_info.get("License"):
             features.license.status = LicenseStatus(license_info.get("status", LicenseStatus.INACTIVE))
             features.license.expired_at = license_info.get("expiredAt", "")
 
-            if workspaces_info := license_info.get("workspaces"):
+            if is_authenticated and (workspaces_info := license_info.get("workspaces")):
                 features.license.workspaces.enabled = workspaces_info.get("enabled", False)
                 features.license.workspaces.limit = workspaces_info.get("limit", 0)
                 features.license.workspaces.size = workspaces_info.get("used", 0)