Repository: apache/airflow
File: task-sdk/src/airflow/sdk/execution_time/task_runner.py (Python)

Code before:
def _check_bundle_permissions_for_impersonation(
    bundle_instance: BaseDagBundle, run_as_user: str, log: structlog.stdlib.BoundLogger
) -> None:
    """
    Check if bundle directories have appropriate permissions for user impersonation.

    When tasks run as a different user via run_as_user, the bundle directories and
    files need to be accessible by that user. This function warns if the permissions
    don't appear to allow group access, which is typically needed for impersonation.

    :param bundle_instance: The bundle instance to check
    :param run_as_user: The user that the task will run as
    :param log: Logger instance for warnings
    """
    import stat

    try:
        bundle_path = bundle_instance.path
        if not bundle_path.exists():
            return

        st = bundle_path.stat()
        mode = st.st_mode

        # Check if group-readable and group-executable (for directories)
        if not (mode & stat.S_IRGRP) or (bundle_path.is_dir() and not (mode & stat.S_IXGRP)):
            log.warning(
                "Bundle path may not be accessible to impersonated user",
                bundle_path=str(bundle_path),
                run_as_user=run_as_user,
                current_permissions=oct(mode),
                hint="Consider setting dag_bundle_new_folder_permissions to 0o775 in "
                "[dag_processor] config, or ensure both users share a common group "
                "with appropriate umask settings (e.g., umask 0002).",
            )
    except Exception:
        # Don't let permission checking failures block task execution
        pass


def startup() -> tuple[RuntimeTaskInstance, Context, Logger]:
    # The parent sends us a StartupDetails message un-prompted. After this, every single message is only sent
    # in response to us sending a request.
    log = structlog.get_logger(logger_name="task")

    if os.environ.get("_AIRFLOW__REEXECUTED_PROCESS") == "1" and (
        msgjson := os.environ.get("_AIRFLOW__STARTUP_MSG")
    ):
        # Clear any Kerberos replace cache if there is one, so new process can't reuse it.
        os.environ.pop("KRB5CCNAME", None)
        # entrypoint of re-exec process

Review comment (security):
Should we just check if we have permissions there? Wile there are some guidelines on how group/ permissions shouldl look like, there are other ways it can be achieved to give those permissions - so I think what we should check here is whether we simply have access and error out if we don't rather than raise warning. In some cases, it might not be desirable to follow those guidelines even if impersonation is used. And in this case users will not be able to get rid of this warning. Instead if we can simply check if we "actually" have acess to those dirs with `os.access()` we can produce a nice error message if we don't.

Code after:
    This is called after user impersonation (if any) to ensure the bundle
    is actually accessible. Uses os.access() which works with any permission
    scheme (standard Unix permissions, ACLs, SELinux, etc.).

    :param bundle_instance: The bundle instance to check
    :param log: Logger instance
    :raises AirflowException: if bundle is not accessible
    """
    from getpass import getuser

    from airflow.exceptions import AirflowException

    bundle_path = bundle_instance.path

    if not bundle_path.exists():
        # Already handled by initialize() with a warning
        return

    # Check read permission (and execute for directories to list contents)
    access_mode = os.R_OK
    if bundle_path.is_dir():
        access_mode |= os.X_OK

    if not os.access(bundle_path, access_mode):
        raise AirflowException(
            f"Bundle '{bundle_instance.name}' path '{bundle_path}' is not accessible "
            f"by user '{getuser()}'. When using run_as_user, ensure bundle directories "
            f"are readable by the impersonated user. "
            f"See: https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/dag-bundles.html"
        )


def startup() -> tuple[RuntimeTaskInstance, Context, Logger]:
    # The parent sends us a StartupDetails message un-prompted. After this, every single message is only sent
    # in response to us sending a request.
    log = structlog.get_logger(logger_name="task")

    if os.environ.get("_AIRFLOW__REEXECUTED_PROCESS") == "1" and (
        msgjson := os.environ.get("_AIRFLOW__STARTUP_MSG")
    ):
        # Clear any Kerberos replace cache if there is one, so new process can't reuse it.
        os.environ.pop("KRB5CCNAME", None)
        # entrypoint of re-exec process

        msg: StartupDetails = TypeAdapter(StartupDetails).validate_json(msgjson)
        reinit_supervisor_comms()

        # We delay this message until _after_ we've got the logging re-configured, otherwise it will show up
        # on stdout
        log.debug("Using serialized startup message from environment", msg=msg)
    else: