Repository: apache/airflow
Pull request: Move Secrets Backend for Variables & Connections support to Task SDK
closes https://github.com/apache/airflow/issues/45435
<!--
 Licensed to the Apache Software Foundation (ASF) under one
 or more contributor license agreements.  See the NOTICE file
 distributed with this work for additional information
 regarding copyright ownership.  The ASF licenses this file
 to you under the Apache License, Version 2.0 (the
 "License"); you may not use this file except in compliance
 with the License.  You may obtain a copy of the License at

   http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing,
 software distributed under the License is distributed on an
 "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 KIND, either express or implied.  See the License for the
 specific language governing permissions and limitations
 under the License.
 -->

<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.

Feel free to ping committers for the review!

In case of an existing issue, reference it using one of the following:

closes: #ISSUE
related: #ISSUE

How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->

## Why?

In addition to retrieving connections & variables from environment variables or the metastore database, you can also enable alternative secrets backend to retrieve Airflow connections or Airflow variables. Now that we have moved the entire logic for Connections, Variables, etc etc, we must support the use case of a task being able to pull in connections and/or variables from their configured backends. Ref: https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/secrets-backend/index.html

Currently, the Task Execution API server handles secrets backend logic, which is ideal for scenarios such as:

> Deploying tasks in transient or ephemeral environments (e.g., GPU cloud services) where secrets need to be retrieved dynamically via the API.

However, there is a complementary use case where secrets or configurations are already provisioned locally on the worker.

> An organization processes sensitive customer data for financial transactions. Regulatory and security policies mandate that certain credentials (e.g., database credentials, API keys, encryption keys) cannot leave a specific network zone. This includes ensuring that the Airflow scheduler, API server, or other components outside this zone cannot access these secrets.

This use-case means we need a way where secrets are already provisioned locally on the worker.


## What needs to be done?

A priority / sequence of backends working should be configured, each of whcih will be tried during the time of retrieval in that sequence.

A sensible order would be defined as:
1. secrets backend on worker directly (optional)
2. env vars on worker
3. reach out to api server
    a. secrets backend defined here (optional)
    b. env vars on api server
    c. metadata DB

Worker will try to get from DB if it doesn't find on ITS own secrets backend and env vars but will do so via the API server.


## Theory worth noting

- Workers will have the chance to configure their own secrets backend now
- The default order in workers would be: 
[configured secrets backend, env var backend on workers, reach out to api server]

- API server as before will have ability to configure its own secrets backend that is unrelated to what the workers have.
- The default order in API server will be:
[configured secrets backend, env var backend on api server, query the metadata DB]


### Config options available

- You can continue using the `AIRFLOW__SECRETS__BACKEND` and related `AIRFLOW__SECRETS__BACKEND_KWARGS` for api server
- Similarly, you can set it for workers now too using: `AIRFLOW__WORKERS__BACKEND` and `AIRFLOW__WORKERS__BACKEND_KWARGS`

^ These are env vars, can also be used via config. Sections are defined here: https://github.com/apache/airflow/pull/47946/files#diff-0a480a8b563e200f749424d7e761555e543588ff39a0b11cce7e34d522f38e0eR1419-R1441


NOTE: I thought this naming culture made the most sense, do let me know if you think there's a better option.


### Example:

```
export AIRFLOW__SECRETS__BACKEND="airflow.sdk.execution_time.apiserver_secrets_backend.APIServerSecretsBackend"
export AIRFLOW__WORKERS__SECRETS_BACKEND ="airflow.secrets.local_filesystem.LocalFilesystemBackend"
export AIRFLOW__WORKERS__SECRETS_BACKEND_KWARGS='{"connections_file_path": "/files/conn.json"}'
```


Config from breeze shell:
```
root@cf562ecaa129:/opt/airflow# airflow config get-value secrets backend
airflow.sdk.execution_time.apiserver_secrets_backend.APIServerSecretsBackend
root@cf562ecaa129:/opt/airflow# airflow config get-value workers backend
airflow.secrets.local_filesystem.LocalFilesystemBackend
```


## Changes of note

### Configuration

#### Config.yaml
- New config introduced for workers --> secrets_backend and secrets_backend_kwargs


#### Configuration Loading
- Leveraging `ensure_secrets_loaded` from earlier, but defined a default to it. The default is the `DEFAULT_SECRETS_SEARCH_PATH` which holds the default priority order for secrets for API server.
- Extended this function so that if its called from worker, with the default values for worker:
```
DEFAULT_SECRETS_SEARCH_PATH_WORKERS = [
    "airflow.secrets.environment_variables.EnvironmentVariablesBackend",
]
```
- It will set the worker mode etc and load the correct custom classes by switiching between "secrets, backend, backend_kwargs" for that of api server vs workers.



### Supervisor
1. We define a global variable: `SECRETS_BACKEND` which is per supervisor process and the reason to do this is because (advantages over defining it in supervisor and not task runner)
- Central place of handling requests to get conn/vars etc
- Caching across dag processes, cache exists but doesnt do much otherwise
- Lesser memory impact, one variable per supervisor instead of one per task, cos there's no real benefit of doing otherwise.

2. Initialised at startup of supervisor or in other words, when a worker fires up


### Definitions

#### Connections

1. The logic for `Connection.get()` is in such a way now that:
```
def _get_connection(conn_id: str) -> Connection:
    from airflow.sdk.execution_time.supervisor import SECRETS_BACKEND
    # TODO: check cache first
    # enabled only if SecretCache.init() has been called first

    # iterate over configured backends if not in cache (or expired)
    for secrets_backend in SECRETS_BACKEND:
        try:
            conn = secrets_backend.get_connection(conn_id=conn_id)
            if conn:
                return conn
        except Exception:
            log.exception(
                "Unable to retrieve connection from secrets backend (%s). "
                "Checking subsequent secrets backend.",
                type(secrets_backend).__name__,
            )

    log.debug(
        "Connection not found in any of the configured Secrets Backends. Trying to retrieve from API server",
        conn_id=conn_id,
    )

    # TODO: This should probably be moved to a separate module like `airflow.sdk.execution_time.comms`
    #   or `airflow.sdk.execution_time.connection`
    #   A reason to not move it to `airflow.sdk.execution_time.comms` is that it
    #   will make that module depend on Task SDK, which is not ideal because we intend to
    #   keep Task SDK as a separate package than execution time mods.
    from airflow.sdk.execution_time.comms import ErrorResponse, GetConnection
    from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS

    SUPERVISOR_COMMS.send_request(log=log, msg=GetConnection(conn_id=conn_id))
    msg = SUPERVISOR_COMMS.get_message()
    if isinstance(msg, ErrorResponse):
        raise AirflowRuntimeError(msg)

    if TYPE_CHECKING:
        assert isinstance(msg, ConnectionResult)
    return _convert_connection_result_conn(msg)
```

TLDR: It tried to look up in all possible configured backends, and if nothing reaches out to the API server.

2. This supports use cases such as:
- User code doing: `Connection.get`
- User code doing: `Basehook.get_connection`
- User code trying to fetch via context

3. For each of the above flow, it will try and find for all backends before reaching out to the DB

#### Variables

- Same as above
```
def _get_variable(key: str, deserialize_json: bool) -> Variable:
    # TODO: check cache first
    # enabled only if SecretCache.init() has been called first
    from airflow.sdk.execution_time.supervisor import SECRETS_BACKEND

    var_val = None
    # iterate over backends if not in cache (or expired)
    for secrets_backend in SECRETS_BACKEND:
        try:
            var_val = secrets_backend.get_variable(key=key)  # type: ignore[assignment]
            if var_val is not None:
                return var_val
        except Exception:
            log.exception(
                "Unable to retrieve variable from secrets backend (%s). "
                "Checking subsequent secrets backend.",
                type(secrets_backend).__name__,
            )

    log.debug(
        "Variable not found in any of the configured Secrets Backends. Trying to retrieve from API server",
        key=key,
    )

    # TODO: This should probably be moved to a separate module like `airflow.sdk.execution_time.comms`
    #   or `airflow.sdk.execution_time.variable`
    #   A reason to not move it to `airflow.sdk.execution_time.comms` is that it
    #   will make that module depend on Task SDK, which is not ideal because we intend to
    #   keep Task SDK as a separate package than execution time mods.
    from airflow.sdk.execution_time.comms import ErrorResponse, GetVariable
    from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS

    SUPERVISOR_COMMS.send_request(log=log, msg=GetVariable(key=key))
    msg = SUPERVISOR_COMMS.get_message()
    if isinstance(msg, ErrorResponse):
        raise AirflowRuntimeError(msg)

    if TYPE_CHECKING:
        assert isinstance(msg, VariableResult)
    return _convert_variable_result_to_variable(msg, deserialize_json)
```


## Testing1: Testing workers priority order

### Setup

- Using breeze
- Create a `files/airflow-breeze-config/init.sh` and add this content to it:

```
export AIRFLOW__SECRETS__BACKEND="airflow.sdk.execution_time.apiserver_secrets_backend.APIServerSecretsBackend"
export AIRFLOW__WORKERS__BACKEND="airflow.secrets.local_filesystem.LocalFilesystemBackend"
export AIRFLOW__WORKERS__BACKEND_KWARGS='{"connections_file_path": "/files/conn.json"}'
```

- Create a conn.json and var.json under /files:

conn.json:

```
{
    "CONN_A": "mysql://host_a",
    "CONN_B": {
        "conn_type": "scheme",
        "host": "host",
        "schema": "schema",
        "login": "Login",
        "password": "None",
        "port": "1234"
    }
}

```

var.json:
```
{
    "VAR_A": "some_value",
    "var_b": "different_value"
}

```


Config from breeze shell after running `breeze start-airflow`
```
root@6c4e36266df8:/opt/airflow# airflow config get-value workers secrets_backend
airflow.secrets.local_filesystem.LocalFilesystemBackend
root@6c4e36266df8:/opt/airflow# airflow config get-value workers secrets_backend_kwargs
{"connections_file_path": "/files/conn.json", "variables_file_path": "/files/var.json"}
```

### Scenario 1: Connection Found in configured secrets backend

DAG:
```
from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.sdk.definitions.connection import Connection

class CustomOperator(BaseOperator):
    def execute(self, context):
        c = Connection.get(conn_id="CONN_A")
        print("The connection is", c)

        print("Connection attributes are", vars(c))


with DAG("example_get_connection_from_secrets", schedule=None, catchup=False) as dag:
    CustomOperator(task_id="print_conn")

```


<img width="1724" alt="image" src="https://github.com/user-attachments/assets/be7a5bfb-e5ed-450c-952f-37187709a72d" />


Logs from worker
```
2025-03-19 14:52:12.255689 [info     ] Connected to redis://redis:6379/0 [celery.worker.consumer.connection]
2025-03-19 14:52:12.260496 [info     ] mingle: searching for neighbors [celery.worker.consumer.mingle]
2025-03-19 14:52:13.266914 [info     ] mingle: all alone              [celery.worker.consumer.mingle]
2025-03-19 14:52:13.275947 [info     ] celery@6c4e36266df8 ready.     [celery.apps.worker]
2025-03-19 14:53:42.702545 [info     ] Task execute_workload[230471aa-482d-4f7f-b48b-f66bd62d0c44] received [celery.worker.strategy]
2025-03-19 14:53:42.861274 [info     ] [230471aa-482d-4f7f-b48b-f66bd62d0c44] Executing workload in Celery: token='eyJhbGciOiJIUzUxMiIsInR5cCI6IkpXVCJ9.eyJhdWQiOiJ1cm46YWlyZmxvdy5hcGFjaGUub3JnOnRhc2siLCJuYmYiOjE3NDIzOTYwMjIsImV4cCI6MTc0MjM5NjYyMiwiaWF0IjoxNzQyMzk2MDIyLCJzdWIiOiIwMTk1YWVlNS01ZDY4LTcyOTUtYTUwZC1mZmVhMjRkYmQyNjUifQ.TABuuDD-0g74gfO2hfAODxXTukzBQn5i9YkiCuBF3X0XABksmMVj6N1rcDzsnW9Kdm79AfkyvBi0OS_VEEgx_A' ti=TaskInstance(id=UUID('0195aee5-5d68-7295-a50d-ffea24dbd265'), task_id='print_conn', dag_id='example_get_connection_from_secrets', run_id='manual__2025-03-19T14:53:42.103685+00:00_0Vi0R8Oo', try_number=1, map_index=-1, pool_slots=1, queue='default', priority_weight=1, executor_config=None) dag_rel_path=PurePosixPath('dags/get_connection.py') bundle_info=BundleInfo(name='dags-folder', version=None) log_path='dag_id=example_get_connection_from_secrets/run_id=manual__2025-03-19T14:53:42.103685+00:00_0Vi0R8Oo/task_id=print_conn/attempt=1.log' kind='ExecuteTask' [airflow.providers.celery.executors.celery_executor_utils]
2025-03-19 14:53:42.885795 [info     ] Initialized secrets backend on workers [supervisor] secrets_backend=[<airflow.secrets.local_filesystem.LocalFilesystemBackend object at 0xffff8c39c940>, <airflow.secrets.environment_variables.EnvironmentVariablesBackend object at 0xffff8c39cb20>]
2025-03-19 14:53:43.582693 [info     ] Task finished                  [supervisor] duration=0.6985896929982118 exit_code=0 final_state=success
2025-03-19 14:53:43.587981 [info     ] Task execute_workload[230471aa-482d-4f7f-b48b-f66bd62d0c44] succeeded in 0.8807627139904071s: None [celery.app.trace]
```



### Scenario 2: Variable Found in configured secrets backend

DAG:
```
from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.sdk.definitions.variable import Variable

class CustomOperator(BaseOperator):
    def execute(self, context):
        v = Variable.get(key="VAR_A")
        print("The variable is", v)


with DAG("example_get_variable_from_secrets", schedule=None, catchup=False, tags=["secrets_backend"]) as dag:
    CustomOperator(task_id="print_var")
```

<img width="1724" alt="image" src="https://github.com/user-attachments/assets/7e3a584d-9dcd-4469-b49d-05f65d618717" />


Task logs:
```
[2025-03-19, 14:56:09] INFO - DAG bundles loaded: dags-folder, example_dags source="airflow.dag_processing.bundles.manager.DagBundlesManager"

[2025-0[3](http://localhost:28080/dags/example_get_variable_from_secrets/runs/manual__2025-03-19T14:56:08.181450+00:00_7u46YbNL/tasks/print_var?try_number=1#3)-19, 14:56:09] INFO - Filling up the DagBag from /files/dags/dags/get_variable.py source="airflow.models.dagbag.DagBag"

[2025-03-19, 1[4](http://localhost:28080/dags/example_get_variable_from_secrets/runs/manual__2025-03-19T14:56:08.181450+00:00_7u46YbNL/tasks/print_var?try_number=1#4):56:09] INFO - Task instance is in running state chan="stdout" source="task"

[2025-03-19, 14:[5](http://localhost:28080/dags/example_get_variable_from_secrets/runs/manual__2025-03-19T14:56:08.181450+00:00_7u46YbNL/tasks/print_var?try_number=1#5)6:09] INFO -  Previous state of the Task instance: queued chan="stdout" source="task"

[2025-03-19, 14:5[6](http://localhost:28080/dags/example_get_variable_from_secrets/runs/manual__2025-03-19T14:56:08.181450+00:00_7u46YbNL/tasks/print_var?try_number=1#6):09] INFO - Current task name:print_var chan="stdout" source="task"

[2025-03-19, 14:56:09] INFO - Dag name:example_get_variable_from_secrets chan="stdout" source="task"

[2025-03-19, 14:56:09] INFO - The variable is some_value chan="stdout" source="task"

[2025-03-1[9](http://localhost:28080/dags/example_get_variable_from_secrets/runs/manual__2025-03-19T14:56:08.181450+00:00_7u46YbNL/tasks/print_var?try_number=1#9), 14:56:09] INFO - Task instance in success state chan="stdout" source="task"

[2025-03-19, 14:56:09] INFO -  Previous state of the Task instance: running chan="stdout" source="task"

[2025-03-19, 14:56:09] INFO - Task operator:<Task(CustomOperator): print_var> chan="stdout" source="task"
```


### Scenario 3: Connection not Found in configured secrets backend, so goes to env vars and is found

- export this env var in worker:
`export AIRFLOW_CONN_AIRFLOW_DB=mysql://root@mysql/airflow?charset=utf8mb4`


DAG:
```
from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.sdk.definitions.connection import Connection

class CustomOperator(BaseOperator):
    def execute(self, context):
        c = Connection.get(conn_id="my_db")
        print("The connection is", c)

        # print("Connection attributes are", vars(c))


with DAG("example_get_connection_from_secrets", schedule=None, catchup=False, tags=["secrets_backend"]) as dag:
    CustomOperator(task_id="print_conn")

```

<img width="1724" alt="image" src="https://github.com/user-attachments/assets/3f7669e7-92f8-4b12-ac8a-1d3895104993" />


Logs:
```
[2025-03-19, 15:00:17] INFO - DAG bundles loaded: dags-folder, example_dags source="airflow.dag_processing.bundles.manager.DagBundlesManager"

[2025-0[3](http://localhost:28080/dags/example_get_connection_from_secrets/runs/manual__2025-03-19T15:00:16.355950+00:00_LtvTKRfT/tasks/print_conn?try_number=1#3)-19, 15:00:17] INFO - Filling up the DagBag from /files/dags/dags/get_connection.py source="airflow.models.dagbag.DagBag"

[2025-03-19, 15:00:17] INFO - Task instance is in running state chan="stdout" source="task"

[202[5](http://localhost:28080/dags/example_get_connection_from_secrets/runs/manual__2025-03-19T15:00:16.355950+00:00_LtvTKRfT/tasks/print_conn?try_number=1#5)-03-19, 15:00:17] INFO -  Previous state of the Task instance: queued chan="stdout" source="task"

[2025-03-19, 15:00:17] INFO - Current task name:print_conn chan="stdout" source="task"

[2025-03-19, 15:00:1[7](http://localhost:28080/dags/example_get_connection_from_secrets/runs/manual__2025-03-19T15:00:16.355950+00:00_LtvTKRfT/tasks/print_conn?try_number=1#7)] INFO - Dag name:example_get_connection_from_secrets chan="stdout" source="task"

[2025-03-19, 15:00:17] INFO - The connection is my_db chan="stdout" source="task"

[2025-03-1[9](http://localhost:28080/dags/example_get_connection_from_secrets/runs/manual__2025-03-19T15:00:16.355950+00:00_LtvTKRfT/tasks/print_conn?try_number=1#9), 15:00:17] INFO - Task instance in success state chan="stdout" source="task"

[2025-03-19, 15:00:17] INFO -  Previous state of the Task instance: running chan="stdout" source="task"

[2025-03-19, 15:00:17] INFO - Task operator:<Task(CustomOperator): print_conn> chan="stdout" source="task"
```



### Scenario 4: Variab;e not Found in configured secrets backend, so goes to env vars and is found

- export this env in your worker:
```
export AIRFLOW_VAR_MY_VAR="my_varvalue"
```

DAG:
```
from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.sdk.definitions.variable import Variable

class CustomOperator(BaseOperator):
    def execute(self, context):
        v = Variable.get(key="MY_VAR")
        print("The variable is", v)


with DAG("example_get_variable_from_secrets", schedule=None, catchup=False, tags=["secrets_backend"]) as dag:
    CustomOperator(task_id="print_var")
```

<img width="1724" alt="image" src="https://github.com/user-attachments/assets/9df63753-5628-43e1-bf0a-d74c96be303b" />


Logs:
```
[2025-03-19, 15:02:31] INFO - DAG bundles loaded: dags-folder, example_dags source="airflow.dag_processing.bundles.manager.DagBundlesManager"

[2025-0[3](http://localhost:28080/dags/example_get_variable_from_secrets/runs/manual__2025-03-19T15:02:20.130858+00:00_ZYTj90V9/tasks/print_var?try_number=1#3)-19, 15:02:31] INFO - Filling up the DagBag from /files/dags/dags/get_variable.py source="airflow.models.dagbag.DagBag"

[2025-03-19, 15:02:31] INFO - Task instance is in running state chan="stdout" source="task"

[202[5](http://localhost:28080/dags/example_get_variable_from_secrets/runs/manual__2025-03-19T15:02:20.130858+00:00_ZYTj90V9/tasks/print_var?try_number=1#5)-03-19, 15:02:31] INFO -  Previous state of the Task instance: queued chan="stdout" source="task"

[2025-03-19, 15:02:31] INFO - Current task name:print_var chan="stdout" source="task"

[2025-03-19, 15:02:31] INFO - Dag name:example_get_variable_from_secrets chan="stdout" source="task"

[2025-03-19, 15:02:31] INFO - The variable is my_varvalue chan="stdout" source="task"

[2025-03-1[9](http://localhost:28080/dags/example_get_variable_from_secrets/runs/manual__2025-03-19T15:02:20.130858+00:00_ZYTj90V9/tasks/print_var?try_number=1#9), 15:02:31] INFO - Task instance in success state chan="stdout" source="task"

[2025-03-19, 15:02:31] INFO -  Previous state of the Task instance: running chan="stdout" source="task"

[2025-03-19, 15:02:31] INFO - Task operator:<Task(CustomOperator): print_var> chan="stdout" source="task"
```

### Scenario 5: Connection not found in any backends, needs to go to API server
- Run breeze with `--load-default-connections` flag

DAG:
```
from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.sdk.definitions.connection import Connection

class CustomOperator(BaseOperator):
    def execute(self, context):
        c = Connection.get(conn_id="athena_default")
        print("The connection is", c)

        # print("Connection attributes are", vars(c))


with DAG("example_get_connection_from_secrets", schedule=None, catchup=False, tags=["secrets_backend"]) as dag:
    CustomOperator(task_id="print_conn")

```

<img width="1724" alt="image" src="https://github.com/user-attachments/assets/2decce79-3d30-4280-a70e-b7f3b7f51b4f" />


API server logs:
![image](https://github.com/user-attachments/assets/62fa873c-2491-4b2f-98bb-5e1faa6aa723)


### Scenario 6: Variable not found in any backends, needs to go to API server

- Create a variable on web UI
<img width="1723" alt="image" src="https://github.com/user-attachments/assets/47e8dbbd-9574-48c1-90f6-7c508b6e38c9" />


DAG:
```
from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.sdk.definitions.variable import Variable

class CustomOperator(BaseOperator):
    def execute(self, context):
        v = Variable.get(key="API_SERVER_VAR")
        print("The variable is", v)


with DAG("example_get_variable_from_secrets", schedule=None, catchup=False, tags=["secrets_backend"]) as dag:
    CustomOperator(task_id="print_var")
```


<img width="1723" alt="image" src="https://github.com/user-attachments/assets/4d2467a9-1df1-48eb-9cc1-9252e728944b" />


API server logs:

![image](https://github.com/user-attachments/assets/1c1816f6-0149-4c01-9e2c-2ddb8bfd6860)


## Testing2: Testing api server priority -- worker backend not configured but secrets backend configured for API server

### Setup:

init.sh:
```
export AIRFLOW__SECRETS_BACKEND="airflow.secrets.local_filesystem.LocalFilesystemBackend"
export AIRFLOW__SECRETS_BACKEND_KWARGS='{"connections_file_path": "/files/conn.json", "variables_file_path": "/files/var.json"}'

```


conn.json:
```
{
    "CONN_API_SERVER_A": "mysql://host_a",
}

```

var.json:
```
{
    "VAR_API_SERVER_A": "some_value",
}
```

Config:
```
root@1630a8cd6e7a:/opt/airflow# airflow config get-value secrets backend
airflow.secrets.local_filesystem.LocalFilesystemBackend
root@1630a8cd6e7a:/opt/airflow# airflow config get-value secrets backend_kwargs
{"connections_file_path": "/files/conn.json", "variables_file_path": "/files/var.json"}
```


### Scenario 1: Connection found in secrets backend of API server
DAG:

```
from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.sdk.definitions.connection import Connection

class CustomOperator(BaseOperator):
    def execute(self, context):
        c = Connection.get(conn_id="CONN_API_SERVER_A")
        print("The connection is", c)

        # print("Connection attributes are", vars(c))


with DAG("example_get_connection_from_secrets", schedule=None, catchup=False, tags=["secrets_backend"]) as dag:
    CustomOperator(task_id="print_conn")

```

![image](https://github.com/user-attachments/assets/3939a140-58bf-4cc6-a5ab-1dc8afbad320)


Call to API server:
<img width="1727" alt="image" src="https://github.com/user-attachments/assets/4cf5b130-636e-4319-b655-f7322ce6132b" />

### Scenario 2: Variable found in secrets backend of API server

DAG:
```
from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.sdk.definitions.variable import Variable

class CustomOperator(BaseOperator):
    def execute(self, context):
        v = Variable.get(key="VAR_API_SERVER_A")
        print("The variable is", v)


with DAG("example_get_variable_from_secrets", schedule=None, catchup=False, tags=["secrets_backend"]) as dag:
    CustomOperator(task_id="print_var")

```

<img width="1727" alt="image" src="https://github.com/user-attachments/assets/7f5133c0-2988-463b-8c46-4dd8f23ce314" />


API server:
![image](https://github.com/user-attachments/assets/c62092ac-3d1e-4c7e-b341-86bcc2fc4fd9)


### Scenario 3: Connection not found in secrets backend of API server, goes to env var and finds it there

Run this in API server:
```
export AIRFLOW_CONN_ENV_VAR_API_SERVER=mysql://root@mysql/airflow?charset=utf8mb4
```

DAG:
```
from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.sdk.definitions.connection import Connection

class CustomOperator(BaseOperator):
    def execute(self, context):
        c = Connection.get(conn_id="ENV_VAR_API_SERVER")
        print("The connection is", c)

        # print("Connection attributes are", vars(c))


with DAG("example_get_connection_from_secrets", schedule=None, catchup=False, tags=["secrets_backend"]) as dag:
    CustomOperator(task_id="print_conn")

```

<img width="1720" alt="image" src="https://github.com/user-attachments/assets/a1a898cf-1350-4a47-b20c-816e9db947af" />

**Same situation for Variables, not trying explicitly.**

### Scenario 4: Connection not found anywhere, call goes to DB

DAG:
```
from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.sdk.definitions.connection import Connection

class CustomOperator(BaseOperator):
    def execute(self, context):
        c = Connection.get(conn_id="ydb_default")
        print("The connection is", c)

        # print("Connection attributes are", vars(c))


with DAG("example_get_connection_from_secrets", schedule=None, catchup=False, tags=["secrets_backend"]) as dag:
    CustomOperator(task_id="print_conn")

```

<img width="1720" alt="image" src="https://github.com/user-attachments/assets/0b9b137f-ee6f-4f8f-a075-231daf755a63" />


![image](https://github.com/user-attachments/assets/f1cea25f-4f2a-4722-9d5d-8785936e4492)

**Same situation for Variables, not trying explicitly.**

### Scenario 5: Connection / variable not found anywhere

DAG:
```
from __future__ import annotations

from airflow.models.baseoperator import BaseOperator
from airflow import DAG
from airflow.sdk.definitions.connection import Connection

class CustomOperator(BaseOperator):
    def execute(self, context):
        c = Connection.get(conn_id="conn_dont_exist")
        print("The connection is", c)

        # print("Connection attributes are", vars(c))


with DAG("example_get_connection_from_secrets", schedule=None, catchup=False, tags=["secrets_backend"]) as dag:
    CustomOperator(task_id="print_conn")

```

<img width="1715" alt="image" src="https://github.com/user-attachments/assets/8792c9a7-67ce-481d-83e0-8c6a5efbcc0c" />




<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).


Problem statement:
Move Secrets Backend for Variables & Connections support to Task SDK
Currently, the Task Execution API server handles secrets backend logic, which is ideal for scenarios such as:

>Deploying tasks in transient or ephemeral environments (e.g., GPU cloud services) where secrets need to be retrieved dynamically via the API.

However, there is a complementary use case where secrets or configurations are already provisioned locally on the worker. 

>An organization processes sensitive customer data for financial transactions. Regulatory and security policies mandate that certain credentials (e.g., database credentials, API keys, encryption keys) cannot leave a specific network zone. This includes ensuring that the Airflow scheduler, API server, or other components outside this zone cannot access these secrets.

This use-case means we need a way where secrets are already provisioned locally on the worker. 

Secrets Backend will be available to be configured on the Client by default.

Longer term the resolution order will be configurable but for now it will be the same as now:
    1. External Secret Backend on worker
    2. Environment Variable on worker
    3. Reach out to API Server

Packaging-wise: The External Secrets Backend will be part of Providers, which depend on Task SDK. The resolution code would be in Task SDK

For the API-Server, by default it will only look at Env Variable and then Database by default.
    1. Optionally, if users want the API server to fetch secrets from the External Secrets backend, they will have to install the Task SDK (and relevant providers like Hashicorp) with the scheduler/Airflow core code.
    2. Longer term: we might be able to remove requirements on Task SDK by implementing some protocol but this will be discussed again post AF 3.0.

Hints from issue discussion:
Do we need to limit worker/client to only env var, or could we allow it to use any secret store if it's configured?

i.e. the full resolution order could be 
- worker env vars
- worker secret backend
- API server env vars
- API server secret backend
- API server database lookup
Look at `get_connection_from_secrets` in airflow.models.connection for where this is used (and you can follow the chain back up to settings and BaseSecretBackend etc.

Commit message: adding support for variables too

Patch to review:
diff --git a/airflow/configuration.py b/airflow/configuration.py
index 239d887bcacbf..3d80752952c97 100644
--- a/airflow/configuration.py
+++ b/airflow/configuration.py
@@ -2026,27 +2026,32 @@ def make_group_other_inaccessible(file_path: str):
         )
 
 
-def ensure_secrets_loaded() -> list[BaseSecretsBackend]:
+def ensure_secrets_loaded(
+    default_backends: list[str] = DEFAULT_SECRETS_SEARCH_PATH,
+) -> list[BaseSecretsBackend]:
     """
     Ensure that all secrets backends are loaded.
 
     If the secrets_backend_list contains only 2 default backends, reload it.
     """
     # Check if the secrets_backend_list contains only 2 default backends
-    if len(secrets_backend_list) == 2:
-        return initialize_secrets_backends()
+    if len(secrets_backend_list) == 2 or default_backends != DEFAULT_SECRETS_SEARCH_PATH:
+        return initialize_secrets_backends(default_backends=default_backends)
     return secrets_backend_list
 
 
-def get_custom_secret_backend() -> BaseSecretsBackend | None:
+def get_custom_secret_backend(worker_mode: bool = False) -> BaseSecretsBackend | None:
     """Get Secret Backend if defined in airflow.cfg."""
-    secrets_backend_cls = conf.getimport(section="secrets", key="backend")
+    section = "workers" if worker_mode else "secrets"
+    key = "backend"
+
+    secrets_backend_cls = conf.getimport(section=section, key=key)
 
     if not secrets_backend_cls:
         return None
 
     try:
-        backend_kwargs = conf.getjson(section="secrets", key="backend_kwargs")
+        backend_kwargs = conf.getjson(section=section, key="backend_kwargs")
         if not backend_kwargs:
             backend_kwargs = {}
         elif not isinstance(backend_kwargs, dict):
@@ -2061,7 +2066,9 @@ def get_custom_secret_backend() -> BaseSecretsBackend | None:
     return secrets_backend_cls(**backend_kwargs)
 
 
-def initialize_secrets_backends() -> list[BaseSecretsBackend]:
+def initialize_secrets_backends(
+    default_backends: list[str] = DEFAULT_SECRETS_SEARCH_PATH,
+) -> list[BaseSecretsBackend]:
     """
     Initialize secrets backend.
 
@@ -2069,13 +2076,16 @@ def initialize_secrets_backends() -> list[BaseSecretsBackend]:
     * instantiate them and return them in a list
     """
     backend_list = []
+    worker_mode = False
+    if default_backends != DEFAULT_SECRETS_SEARCH_PATH:
+        worker_mode = True
 
-    custom_secret_backend = get_custom_secret_backend()
+    custom_secret_backend = get_custom_secret_backend(worker_mode)
 
     if custom_secret_backend is not None:
         backend_list.append(custom_secret_backend)
 
-    for class_name in DEFAULT_SECRETS_SEARCH_PATH:
+    for class_name in default_backends:
         secrets_backend_cls = import_string(class_name)
         backend_list.append(secrets_backend_cls())
 
diff --git a/airflow/secrets/__init__.py b/airflow/secrets/__init__.py
index e59c1f5e74ca0..b502f1385f8b0 100644
--- a/airflow/secrets/__init__.py
+++ b/airflow/secrets/__init__.py
@@ -27,7 +27,7 @@
 
 from __future__ import annotations
 
-__all__ = ["BaseSecretsBackend", "DEFAULT_SECRETS_SEARCH_PATH"]
+__all__ = ["BaseSecretsBackend", "DEFAULT_SECRETS_SEARCH_PATH", "DEFAULT_SECRETS_SEARCH_PATH_WORKERS"]
 
 from airflow.secrets.base_secrets import BaseSecretsBackend
 
@@ -35,3 +35,7 @@
     "airflow.secrets.environment_variables.EnvironmentVariablesBackend",
     "airflow.secrets.metastore.MetastoreBackend",
 ]
+
+DEFAULT_SECRETS_SEARCH_PATH_WORKERS = [
+    "airflow.secrets.environment_variables.EnvironmentVariablesBackend",
+]
diff --git a/task-sdk/src/airflow/sdk/definitions/connection.py b/task-sdk/src/airflow/sdk/definitions/connection.py
index 5a447895679ff..ccc96593e870d 100644
--- a/task-sdk/src/airflow/sdk/definitions/connection.py
+++ b/task-sdk/src/airflow/sdk/definitions/connection.py
@@ -24,7 +24,8 @@
 
 import attrs
 
-from airflow.exceptions import AirflowException
+from airflow.exceptions import AirflowException, AirflowNotFoundException
+from airflow.sdk.api.datamodels._generated import ConnectionResponse
 
 log = logging.getLogger(__name__)
 
@@ -87,6 +88,35 @@ def get(cls, conn_id: str) -> Any:
 
         return _get_connection(conn_id)
 
+    @classmethod
+    def get_connection_from_secrets(cls, conn_id: str) -> Connection:
+        """
+        Get connection by conn_id by iterating over all Secret Backends.
+
+        :param conn_id: connection id
+        :return: connection
+        """
+        # TODO: check cache first
+        # enabled only if SecretCache.init() has been called first
+
+        # iterate over configured backends if not in cache (or expired)
+        from airflow.sdk.execution_time.supervisor import SECRETS_BACKEND
+
+        for secrets_backend in SECRETS_BACKEND:
+            print("The backend in the sdk is", secrets_backend)
+            try:
+                conn = secrets_backend.get_connection(conn_id=conn_id)
+                if conn:
+                    return conn
+            except Exception:
+                log.exception(
+                    "Unable to retrieve connection from secrets backend (%s). "
+                    "Checking subsequent secrets backend.",
+                    type(secrets_backend).__name__,
+                )
+
+        raise AirflowNotFoundException(f"The conn_id `{conn_id}` isn't defined")
+
     @property
     def extra_dejson(self) -> dict:
         """Deserialize `extra` property to JSON."""
@@ -98,3 +128,15 @@ def extra_dejson(self) -> dict:
                 log.exception("Failed to deserialize extra property `extra`, returning empty dictionary")
         # TODO: Mask sensitive keys from this list or revisit if it will be done in server
         return extra
+
+    def _convert_connection_to_response(self) -> ConnectionResponse:
+        return ConnectionResponse(
+            conn_id=self.conn_id,
+            conn_type=self.conn_type,
+            host=self.host,
+            schema_=self.schema,
+            login=self.login,
+            password=self.password,
+            port=self.port,
+            extra=self.extra,
+        )
diff --git a/task-sdk/src/airflow/sdk/definitions/variable.py b/task-sdk/src/airflow/sdk/definitions/variable.py
index 46742e965ed5e..86e6a8366d749 100644
--- a/task-sdk/src/airflow/sdk/definitions/variable.py
+++ b/task-sdk/src/airflow/sdk/definitions/variable.py
@@ -17,12 +17,17 @@
 
 from __future__ import annotations
 
+import logging
 from typing import Any
 
 import attrs
 
+from airflow.exceptions import AirflowNotFoundException
+from airflow.sdk.api.datamodels._generated import VariableResponse
 from airflow.sdk.definitions._internal.types import NOTSET
 
+log = logging.getLogger(__name__)
+
 
 @attrs.define
 class Variable:
@@ -52,3 +57,38 @@ def get(cls, key: str, default: Any = NOTSET, deserialize_json: bool = False):
             if e.error.error == ErrorType.VARIABLE_NOT_FOUND and default is not NOTSET:
                 return default
             raise
+
+    @classmethod
+    def get_variable_from_secrets(cls, key: str):
+        """
+        Get Airflow Variable by iterating over all Secret Backends.
+
+        :param key: Variable Key
+        :return: Variable Value
+        """
+        # TODO: check cache first
+        # enabled only if SecretCache.init() has been called first
+        from airflow.sdk.execution_time.supervisor import SECRETS_BACKEND
+
+        var_val = NOTSET
+        # iterate over backends if not in cache (or expired)
+        for secrets_backend in SECRETS_BACKEND:
+            try:
+                var_val = secrets_backend.get_variable(key=key)
+                if var_val is not NOTSET:
+                    return var_val
+            except Exception:
+                log.exception(
+                    "Unable to retrieve variable from secrets backend (%s). "
+                    "Checking subsequent secrets backend.",
+                    type(secrets_backend).__name__,
+                )
+
+        if var_val is not NOTSET:
+            raise AirflowNotFoundException(f"The variable with key `{key}` isn't defined")
+
+        return var_val
+
+    @staticmethod
+    def _convert_variable_to_response(key: str, value: str | None) -> VariableResponse:
+        return VariableResponse(key=key, value=value)
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index b73db7d568f04..1da6cf7e7d606 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -34,6 +34,7 @@
 from socket import SocketIO, socket, socketpair
 from typing import (
     TYPE_CHECKING,
+    Any,
     BinaryIO,
     Callable,
     ClassVar,
@@ -52,6 +53,8 @@
 import structlog
 from pydantic import TypeAdapter
 
+from airflow.exceptions import AirflowNotFoundException
+from airflow.sdk import Connection, Variable
 from airflow.sdk.api.client import Client, ServerResponseError
 from airflow.sdk.api.datamodels._generated import (
     AssetResponse,
@@ -102,7 +105,7 @@
     from airflow.typing_compat import Self
 
 
-__all__ = ["ActivitySubprocess", "WatchedSubprocess", "supervise"]
+__all__ = ["ActivitySubprocess", "WatchedSubprocess", "supervise", "SECRETS_BACKEND"]
 
 log: FilteringBoundLogger = structlog.get_logger(logger_name="supervisor")
 
@@ -123,6 +126,8 @@
     TerminalTIState.SUCCESS,
 ]
 
+SECRETS_BACKEND: Any
+
 
 @overload
 def mkpipe() -> tuple[socket, socket]: ...
@@ -873,14 +878,24 @@ def _handle_request(self, msg: ToSupervisor, log: FilteringBoundLogger):
                 outlet_events=msg.outlet_events,
             )
         elif isinstance(msg, GetConnection):
-            conn = self.client.connections.get(msg.conn_id)
+            try:
+                connection = Connection.get_connection_from_secrets(msg.conn_id)
+                conn = connection._convert_connection_to_response()
+            except AirflowNotFoundException:
+                # if none of the backend has the connection defined, use the API to check instead
+                conn = self.client.connections.get(msg.conn_id)
             if isinstance(conn, ConnectionResponse):
                 conn_result = ConnectionResult.from_conn_response(conn)
                 resp = conn_result.model_dump_json(exclude_unset=True, by_alias=True).encode()
             else:
                 resp = conn.model_dump_json().encode()
         elif isinstance(msg, GetVariable):
-            var = self.client.variables.get(msg.key)
+            try:
+                variable_value = Variable.get_variable_from_secrets(msg.key)
+                var = Variable._convert_variable_to_response(msg.key, variable_value)
+            except AirflowNotFoundException:
+                # if none of the backend has the variable defined, use the API to check instead
+                var = self.client.variables.get(msg.key)
             if isinstance(var, VariableResponse):
                 var_result = VariableResult.from_variable_response(var)
                 resp = var_result.model_dump_json(exclude_unset=True).encode()
@@ -1051,6 +1066,15 @@ def forward_to_log(
             log.log(level, msg, chan=chan)
 
 
+def initialize_secrets_backend_on_workers():
+    """Initialize the secrets backend on workers."""
+    from airflow.configuration import DEFAULT_SECRETS_SEARCH_PATH_WORKERS, ensure_secrets_loaded
+
+    global SECRETS_BACKEND
+    SECRETS_BACKEND = ensure_secrets_loaded(default_backends=DEFAULT_SECRETS_SEARCH_PATH_WORKERS)
+    log.info("The secrets backend is", secrets_backend=SECRETS_BACKEND)
+
+
 def supervise(
     *,
     ti: TaskInstance,
@@ -1108,6 +1132,8 @@ def supervise(
         processors = logging_processors(enable_pretty_log=pretty_logs)[0]
         logger = structlog.wrap_logger(underlying_logger, processors=processors, logger_name="task").bind()
 
+    initialize_secrets_backend_on_workers()
+
     process = ActivitySubprocess.start(
         dag_rel_path=dag_rel_path,
         what=ti,


Code review comments:
diff_hunk: @@ -52,3 +57,38 @@ def get(cls, key: str, default: Any = NOTSET, deserialize_json: bool = False):
             if e.error.error == ErrorType.VARIABLE_NOT_FOUND and default is not NOTSET:
                 return default
             raise
+
+    @classmethod
+    def get_variable_from_secrets(cls, key: str):
+        """
+        Get Airflow Variable by iterating over all Secret Backends.
+
+        :param key: Variable Key
+        :return: Variable Value
+        """
+        # TODO: check cache first
+        # enabled only if SecretCache.init() has been called first
+        from airflow.sdk.execution_time.supervisor import SECRETS_BACKEND
+
+        var_val = NOTSET
+        # iterate over backends if not in cache (or expired)
+        for secrets_backend in SECRETS_BACKEND:
+            try:
+                var_val = secrets_backend.get_variable(key=key)
+                if var_val is not NOTSET:
+                    return var_val
+            except Exception:
+                log.exception(
+                    "Unable to retrieve variable from secrets backend (%s). "
+                    "Checking subsequent secrets backend.",
+                    type(secrets_backend).__name__,
+                )
+
+        if var_val is not NOTSET:
+            raise AirflowNotFoundException(f"The variable with key `{key}` isn't defined")
+
+        return var_val
+
+    @staticmethod
+    def _convert_variable_to_response(key: str, value: str | None) -> VariableResponse:
+        return VariableResponse(key=key, value=value)
path: task-sdk/src/airflow/sdk/definitions/variable.py
text: @user1:
One option is to move this entire thing including the sequencing in supervisor.py to `airflow.sdk.execution_time.context. _get_variable` function -- so it becomes a single way to get variable internally.

@author:
I am taking this path ahead and proceeding

@author:
I did this in latest commit: b7ad2e6da6cfa0d08d3c0d3209162452e7d4d851

diff_hunk: @@ -123,6 +126,8 @@
     TerminalTIState.SUCCESS,
 ]
 
+SECRETS_BACKEND: Any
path: task-sdk/src/airflow/sdk/execution_time/supervisor.py
text: @user1:
Shouldn't be any!

@author:
Right yeah, set it to the correct type

diff_hunk: @@ -24,7 +24,8 @@
 
 import attrs
 
-from airflow.exceptions import AirflowException
+from airflow.exceptions import AirflowException, AirflowNotFoundException
+from airflow.sdk.api.datamodels._generated import ConnectionResponse
path: task-sdk/src/airflow/sdk/definitions/connection.py
text: @user1:
Since this is a user facing class, let's avoid importing `ConnectionResponse` and defining `_convert_connection_to_response` here if at all possible.



@author:
It is possible yes, let me try and move it to supervisor itself im thinking. It doesnt have to be linked to the class too, its a general util method

@author:
But that doesn't feel too right too. 

@author:
I moved those utils to supervisor for the time being, LMK what you think