Repository: apache/airflow
File: providers/fab/src/airflow/providers/fab/auth_manager/fab_auth_manager.py (Python)

Code before:
    AccessView.CLUSTER_ACTIVITY: RESOURCE_CLUSTER_ACTIVITY,
    AccessView.DOCS: RESOURCE_DOCS,
    AccessView.IMPORT_ERRORS: RESOURCE_IMPORT_ERROR,
    AccessView.JOBS: RESOURCE_JOB,
    AccessView.PLUGINS: RESOURCE_PLUGIN,
    AccessView.PROVIDERS: RESOURCE_PROVIDER,
    AccessView.TRIGGERS: RESOURCE_TRIGGER,
    AccessView.WEBSITE: RESOURCE_WEBSITE,
}

_MAP_MENU_ITEM_TO_FAB_RESOURCE_TYPE = {
    MenuItem.ASSETS: RESOURCE_ASSET,
    MenuItem.AUDIT_LOG: RESOURCE_AUDIT_LOG,
    MenuItem.CONFIG: RESOURCE_CONFIG,
    MenuItem.CONNECTIONS: RESOURCE_CONNECTION,
    MenuItem.DAGS: RESOURCE_DAG,
    MenuItem.DOCS: RESOURCE_DOCS,
    MenuItem.PLUGINS: RESOURCE_PLUGIN,
    MenuItem.POOLS: RESOURCE_POOL,
    MenuItem.PROVIDERS: RESOURCE_PROVIDER,
    MenuItem.VARIABLES: RESOURCE_VARIABLE,
    MenuItem.XCOMS: RESOURCE_XCOM,
}


if AIRFLOW_V_3_1_PLUS:
    from airflow.providers.fab.www.security.permissions import RESOURCE_HITL_DETAIL

    _MAP_MENU_ITEM_TO_FAB_RESOURCE_TYPE[MenuItem.REQUIRED_ACTIONS] = RESOURCE_HITL_DETAIL
    _MAP_DAG_ACCESS_ENTITY_TO_FAB_RESOURCE_TYPE[DagAccessEntity.HITL_DETAIL] = (RESOURCE_HITL_DETAIL,)


class FabAuthManager(BaseAuthManager[User]):
    """
    Flask-AppBuilder auth manager.

    This auth manager is responsible for providing a backward compatible user management experience to users.
    """

    appbuilder: AirflowAppBuilder | None = None

    def init_flask_resources(self) -> None:
        self._sync_appbuilder_roles()

    @cached_property
    def apiserver_endpoint(self) -> str:
        return conf.get("api", "base_url", fallback="/")

    @staticmethod
    def get_cli_commands() -> list[CLICommand]:
        """Vends CLI commands to be included in Airflow CLI."""

Review comment (security):
Have you tried to update this value in the config? By setting this constant at the module level I am not sure Airflow has time to load config from providers before loading this file. Instead of a constant at the module level, I would create a property with `@cached_property` in `FabAuthManager`

Code after:
_MAP_ACCESS_VIEW_TO_FAB_RESOURCE_TYPE = {
    AccessView.CLUSTER_ACTIVITY: RESOURCE_CLUSTER_ACTIVITY,
    AccessView.DOCS: RESOURCE_DOCS,
    AccessView.IMPORT_ERRORS: RESOURCE_IMPORT_ERROR,
    AccessView.JOBS: RESOURCE_JOB,
    AccessView.PLUGINS: RESOURCE_PLUGIN,
    AccessView.PROVIDERS: RESOURCE_PROVIDER,
    AccessView.TRIGGERS: RESOURCE_TRIGGER,
    AccessView.WEBSITE: RESOURCE_WEBSITE,
}

_MAP_MENU_ITEM_TO_FAB_RESOURCE_TYPE = {
    MenuItem.ASSETS: RESOURCE_ASSET,
    MenuItem.AUDIT_LOG: RESOURCE_AUDIT_LOG,
    MenuItem.CONFIG: RESOURCE_CONFIG,
    MenuItem.CONNECTIONS: RESOURCE_CONNECTION,
    MenuItem.DAGS: RESOURCE_DAG,
    MenuItem.DOCS: RESOURCE_DOCS,
    MenuItem.PLUGINS: RESOURCE_PLUGIN,
    MenuItem.POOLS: RESOURCE_POOL,
    MenuItem.PROVIDERS: RESOURCE_PROVIDER,
    MenuItem.VARIABLES: RESOURCE_VARIABLE,
    MenuItem.XCOMS: RESOURCE_XCOM,
}

CACHE_TTL = conf.getint("fab", "cache_ttl", fallback=30)

if AIRFLOW_V_3_1_PLUS:
    from airflow.providers.fab.www.security.permissions import RESOURCE_HITL_DETAIL

    _MAP_MENU_ITEM_TO_FAB_RESOURCE_TYPE[MenuItem.REQUIRED_ACTIONS] = RESOURCE_HITL_DETAIL
    _MAP_DAG_ACCESS_ENTITY_TO_FAB_RESOURCE_TYPE[DagAccessEntity.HITL_DETAIL] = (RESOURCE_HITL_DETAIL,)


class FabAuthManager(BaseAuthManager[User]):
    """
    Flask-AppBuilder auth manager.

    This auth manager is responsible for providing a backward compatible user management experience to users.
    """

    cache: TTLCache = TTLCache(maxsize=1024, ttl=CACHE_TTL)
    appbuilder: AirflowAppBuilder | None = None

    def init_flask_resources(self) -> None:
        self._sync_appbuilder_roles()

    @cached_property
    def apiserver_endpoint(self) -> str:
        return conf.get("api", "base_url", fallback="/")