Repository: apache/airflow
Pull request: AIP-72: Improving Operator Links Interface to Prevent User Code Execution in Webserver
<!--
 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/
-->

closes: https://github.com/apache/airflow/issues/46609

## Why?
If you want to add further links to operators you can define them via a plugin or provider package. Extra links will be displayed in task details page in Grid view.

Currently, users define operator links in one of two ways:
1. Using plugins and defining operators in those plugins for defining extra links 
2. Directly using custom airflow operators to define operator extra links.

This feature doesn't work with Airflow 3, so while porting this to task sdk so it can be used by users, we decided to simplify the extra links code-flow in a way to also reduce / decouple the user code that runs using `get_link` of a operator extra link in the airflow webserver to outside the context of the webserver.


## What?
To get this working, a simple interface change is made and the following things get easier:
1. While defining an extra link, add a "xcom_key" required field to it.

For example:
```
class FooBarLink(BaseOperatorLink):
    name = "foo-bar"
    xcom_key = "link-key"

    def get_link(self, operator, *, ti_key):
        return f"http://www.example.com"
```

2. At runtime, from the task runner, after the task runs (when its `finalize`ing), it requests to push an XCOM which contains the "entire" link for the extra link with the xcom key and task id.

Ex:
```
def finalize(ti: RuntimeTaskInstance, log: Logger):
    for oe in ti.task.operator_extra_link_dict.values():
        link, xcom_key = oe.get_link(operator=ti.task, ti_key=ti.id), oe.xcom_key  # type: ignore[arg-type]
        log.debug("Setting xcom for operator extra link", link=link, xcom_key=xcom_key)
        _xcom_push(ti, key=xcom_key, value=link)
```

3. After the task has completed running, when the user goes to grid view, he can see the extra button.
The link in this button comes by reading from xcom instead of running user code

## How?

### Changes of note:

#### BaseOperatorLink Model
Introducing a new `xcom_key` datafield that carries the xcom_key with which the entire link will be pushed.

Introduced a new slim interface that we will use during deserialisation which can return the link from xcoms instead of user defined "get_link".
```
@attrs.define()
class GenericOperatorLink(LoggingMixin):
    """A generic operator link class that can retrieve link only using XCOMs. Used while deserializing operators."""

    name: str
    xcom_key: str

    def get_link(self, ti: TaskInstance) -> str:
        """
        Retrieve the link from the XComs.
        :param ti: Task instance from which to retrieve the link
        :return: link to external system, but by pulling it from XComs
        """
        self.log.info("Retrieving link from XComs with key: %s for task id: %s", self.xcom_key, ti.task_id)
        return ti.xcom_pull(key=self.xcom_key, task_ids=ti.task_id)  # type: ignore
```


#### DAG serialisation/ deserialisation
We need something in the serdag to say what links a task has, which we already have but the format is overly complex.

Difference in formats (then vs now):
<meta charset="utf-8"><b style="font-weight:normal;" id="docs-internal-guid-aeba09dd-7fff-6e38-86db-19bab8077e4b"><br /><br /><div dir="ltr" style="margin-left:0pt;" align="left">
  | Older | Newer
-- | -- | --
Encoded Operator | '_operator_extra_links': [{'tests.www.views.test_views_extra_links.RaiseErrorLink': {}},  {'tests.www.views.test_views_extra_links.NoResponseLink': {}},  {'tests.www.views.test_views_extra_links.FooBarLink': {}},  {'tests_common.test_utils.mock_operators.AirflowLink': {}}]} | '_operator_extra_links': {'raise_error': 'key',  'no_response': 'key',  'foo-bar': 'link-key',  'airflow': 'airflow_link_key'}}
Decoded Operator | _operator_links_source ={   'airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleIndexableLink': {       'index': 0   }},list(_operator_links_source.items()) =[   (       'airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleIndexableLink',       {'index': 0}   )]list(_operator_links_source.items())[0] =(   'airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleIndexableLink',   {       'index': 0   }) | deserialised_operator = SerializedBaseOperator.deserialize_operator(enc_operator)deserialised_operator.operator_extra_linksOut[8]: [GenericOperatorLink(name='raise_error', xcom_key='key'), GenericOperatorLink(name='no_response', xcom_key='key'), GenericOperatorLink(name='foo-bar', xcom_key='link-key'), GenericOperatorLink(name='airflow', xcom_key='airflow_link_key')]

</div><br /></b>

Example of how its stored in the serialised dag now:
```
{"__version": 1, "dag": {"tags": ["extra_links"], "edge_info": {}, "task_group": {"_group_id": null, "group_display_name": "", "prefix_group_id": true, "tooltip": "", "ui_color": "CornflowerBlue", "ui_fgcolor": "#000", "children": {"task": ["operator", "task"]}, "upstream_group_ids": [], "downstream_group_ids": [], "upstream_task_ids": [], "downstream_task_ids": []}, "timetable": {"__type": "airflow.timetables.simple.NullTimetable", "__var": {}}, "fileloc": "/files/dags/dags/extra_links_custom_operator.py", "dag_id": "extra_links_custom_operator", "timezone": "UTC", "catchup": false, "relative_fileloc": "dags/extra_links_custom_operator.py", "_processor_dags_folder": "/files/dags", "tasks": [{"__var": {"pool": "default_pool", "task_type": "DummyTestOperator", "downstream_task_ids": [], "template_fields": [], "on_failure_fail_dagrun": false, "template_fields_renderers": {}, "task_id": "task", "is_teardown": false, "ui_color": "#fff", "weight_rule": "downstream", "ui_fgcolor": "#000", "start_from_trigger": false, "_needs_expansion": false, "template_ext": [], "is_setup": false, "_task_module": "unusual_prefix_86147154bd7e391eb5735fc97db99fe835bcf5b0_extra_links_custom_operator", "_is_empty": false, "start_trigger_args": null, "_operator_extra_links": {"no_response": "key", "foo-bar": "link-key"}, "label": "task", "_operator_name": "DummyTestOperator"}, "__type": "operator"}], "dag_dependencies": [], "params": []}}
```

- During deserialisation, we create a new GenericOperatorLink type link for each link defined per operator in the serdag

#### Abstract Operator

- The definitions for utils for operator extra links have been moved to task sdk.
- The models/abstractoperator inherits from the task sdk abstarctoperator.
- The `get_extra_links` in task sdk now follows the new function signature

## Testing

1. Operator Links Defined at Operator Level

DAG:
```
from airflow.models import BaseOperatorLink, BaseOperator
from airflow.sdk import DAG

class NoResponseLink(BaseOperatorLink):
    name = "no_response"
    xcom_key = "key"

    def get_link(self, operator, *, ti_key):
        return None


class FooBarLink(BaseOperatorLink):
    name = "foo-bar"
    xcom_key = "link-key"

    def get_link(self, operator, *, ti_key):
        return f"http://www.example.com"


class DummyTestOperator(BaseOperator):
    operator_extra_links = (
        NoResponseLink(),
        FooBarLink(),
    )

    def execute(self, context):
        print("Hello from custom operator", self.operator_extra_links)

with DAG(
    dag_id="extra_links_custom_operator",
    schedule=None,
    catchup=False,
    tags=["extra_links"],
):

    t = DummyTestOperator(
        task_id="task"
    )

```

![image](https://github.com/user-attachments/assets/fbe502d1-ad5b-46e3-b323-0852a1be002b)


The link elements:
```
<a target="_blank" class="chakra-link chakra-button c-h1717v" href="http://www.example.com">foo-bar</a>
<a class="chakra-link chakra-button c-h1717v" href="null">no_response</a>
```

XCOMS pushed:
<img width="1471" alt="image" src="https://github.com/user-attachments/assets/bf3bfcc4-485c-473d-922b-e7aa2fe31ad8" />

Logs:
<img width="1471" alt="image" src="https://github.com/user-attachments/assets/dd4b8747-be43-47c7-82e5-9e72d210d55b" />

Last few lines:
```
{"timestamp":"2025-02-10T09:15:06.800071","level":"debug","event":"Sending request","json":"{\"state\":\"success\",\"end_date\":\"2025-02-10T09:15:06.800015Z\",\"task_outlets\":[],\"outlet_events\":[],\"type\":\"SucceedTask\"}\n","logger":"task"}
{"timestamp":"2025-02-10T09:15:06.800145","level":"debug","event":"Plugins are already loaded. Skipping.","logger":"airflow.plugins_manager"}
{"timestamp":"2025-02-10T09:15:06.800175","level":"debug","event":"Initialize extra operators links plugins","logger":"airflow.plugins_manager"}
{"timestamp":"2025-02-10T09:15:06.800263","level":"debug","event":"Setting xcom for operator extra link","link":null,"xcom_key":"key","logger":"task"}
{"timestamp":"2025-02-10T09:15:06.800411","level":"debug","event":"Sending request","json":"{\"key\":\"key\",\"value\":\"null\",\"dag_id\":\"extra_links_custom_operator\",\"run_id\":\"manual__2025-02-10T09:15:05.441050+00:00\",\"task_id\":\"task\",\"map_index\":-1,\"mapped_length\":null,\"type\":\"SetXCom\"}\n","logger":"task"}
{"timestamp":"2025-02-10T09:15:06.800430","level":"debug","event":"Setting xcom for operator extra link","link":"http://www.example.com/","xcom_key":"link-key","logger":"task"}
{"timestamp":"2025-02-10T09:15:06.800457","level":"debug","event":"Sending request","json":"{\"key\":\"link-key\",\"value\":\"\\\"http://www.example.com/\\\"\",\"dag_id\":\"extra_links_custom_operator\",\"run_id\":\"manual__2025-02-10T09:15:05.441050+00:00\",\"task_id\":\"task\",\"map_index\":-1,\"mapped_length\":null,\"type\":\"SetXCom\"}\n","logger":"task"}
```


2. Operator Links Defined as plugins
Plugin:
```
from airflow.plugins_manager import AirflowPlugin
from airflow.models.baseoperatorlink import BaseOperatorLink
from airflow.providers.standard.operators.python import PythonOperator


# define the extra link
class HTTPDocsLink(BaseOperatorLink):
    # name the link button in the UI
    name = "HTTP docs"
    xcom_key = "key"

    # add the button to one or more operators
    operators = [PythonOperator]

    # provide the link
    def get_link(self, operator, *, ti_key=None):
        return "https://developer.mozilla.org/en-US/docs/Web/HTTP"

# define the plugin class
class AirflowExtraLinkPlugin(AirflowPlugin):
    name = "extra_link_plugin"
    operator_extra_links = [
        HTTPDocsLink(),
    ]

```

DAG:
```
from airflow.models.dag import DAG
from pendulum import datetime

from airflow.providers.standard.operators.python import PythonOperator


def handler():
    print("Hello from the python operator!!")

with DAG(
    dag_id="extra_links_plugin",
    start_date=datetime(2022, 11, 1),
    schedule=None,
    catchup=False,
    tags=["extra_links"],
):

    call_api_simple = PythonOperator(
        task_id="call_api_simple",
        python_callable=handler,
    )

```


<img width="1471" alt="image" src="https://github.com/user-attachments/assets/89bf6e7b-58d9-4801-a8f3-35f7b258e78b" />

The link elements:
```
<a target="_blank" class="chakra-link chakra-button c-h1717v" href="https://developer.mozilla.org/en-US/docs/Web/HTTP">HTTP docs</a>
```

XComs pushed:
<img width="1471" alt="image" src="https://github.com/user-attachments/assets/615f195d-8a63-4d7a-9f80-c6893631f2ba" />

Logs:
<img width="1471" alt="image" src="https://github.com/user-attachments/assets/96f8d4ba-eee6-4ec0-8ed9-adf6efab8b90" />

Last few logs lines:
```
{"timestamp":"2025-02-10T09:19:37.131169","level":"debug","event":"Sending request","json":"{\"state\":\"success\",\"end_date\":\"2025-02-10T09:19:37.131068Z\",\"task_outlets\":[],\"outlet_events\":[],\"type\":\"SucceedTask\"}\n","logger":"task"}
{"timestamp":"2025-02-10T09:19:37.131332","level":"debug","event":"Plugins are already loaded. Skipping.","logger":"airflow.plugins_manager"}
{"timestamp":"2025-02-10T09:19:37.131368","level":"debug","event":"Initialize extra operators links plugins","logger":"airflow.plugins_manager"}
{"timestamp":"2025-02-10T09:19:37.131524","level":"debug","event":"Setting xcom for operator extra link","link":"[https://developer.mozilla.org/en-US/docs/Web/HTTP","xcom_key":"key","logger":"task"](https://developer.mozilla.org/en-US/docs/Web/HTTP%22,%22xcom_key%22:%22key%22,%22logger%22:%22task%22)}
{"timestamp":"2025-02-10T09:19:37.131681","level":"debug","event":"Sending request","json":"{\"key\":\"key\",\"value\":\"\\\"https://developer.mozilla.org/en-US/docs/Web/HTTP\\\"\",\"dag_id\":\"extra_links_plugin\",\"run_id\":\"manual__2025-02-10T09:19:36.380827+00:00\",\"task_id\":\"call_api_simple\",\"map_index\":-1,\"mapped_length\":null,\"type\":\"SetXCom\"}\n","logger":"task"}
{"timestamp":"2025-02-10T09:19:37.877771Z","level":"info","event":"Hello from the python operator!!","chan":"stdout","logger":"task"}
```

## Whats pending?
- [x] Adding newsfragments (significant / behaviour change?)
- [ ] Testing with SDK baseoperator
- [ ] Moving baseoperatorlink model to sdk definitions

<!-- 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:
Get Operator Extra Links working again
Operator links do not work in Airflow 3.

- The plan is to change the interface in a way that webserver doesn't have to execute the user code: `get_link`
- We will push an xcom during task runtime to the XCOM backend. The key will be as defined in the operator link
- While reading from the webserver, instead of reading from the `get_link` method, XCom is read.

Hints from issue discussion:
>We will push an xcom during task runtime to the XCOM backend. The key will be as defined in the operator link

How is this going to work when user created several extra links for the same operator?
We will be adding "xcom_key" at the interface level for operator links. For example:
```
class DummyTestOperator(BaseOperator):
    # We need to ignore type check here due to 2.7 compatibility import
    operator_extra_links = (  # type: ignore[assignment]
        RaiseErrorLink(),
        NoResponseLink(),
        FooBarLink(),
        AirflowLink(),
    )
```

For such cases, each of the links here are attached to this operator and they have XCOM keys of their own.
So while doing a lookup from webserver, we will fetch all the links pertaining to the operator.

We also are storing at the serdag level to assist with this

Commit message: harder review comments: moved the get_extra_links to serialisedBaseOperator, some refactoring, signature of GOL to BOL

Patch to review:
diff --git a/airflow/models/abstractoperator.py b/airflow/models/abstractoperator.py
index 98fd977c59128..e1d909faa87bd 100644
--- a/airflow/models/abstractoperator.py
+++ b/airflow/models/abstractoperator.py
@@ -18,15 +18,12 @@
 from __future__ import annotations
 
 import datetime
-import inspect
 from collections.abc import Iterable, Sequence
-from functools import cached_property
 from typing import TYPE_CHECKING, Any, Callable
 
 from sqlalchemy import select
 
 from airflow.configuration import conf
-from airflow.exceptions import AirflowException
 from airflow.sdk.definitions._internal.abstractoperator import (
     AbstractOperator as TaskSDKAbstractOperator,
     NotMapped as NotMapped,  # Re-export this for compat
@@ -42,7 +39,6 @@
 if TYPE_CHECKING:
     from sqlalchemy.orm import Session
 
-    from airflow.models.baseoperatorlink import BaseOperatorLink
     from airflow.models.dag import DAG as SchedulerDAG
     from airflow.models.taskinstance import TaskInstance
     from airflow.sdk.definitions.baseoperator import BaseOperator
@@ -157,64 +153,6 @@ def priority_weight_total(self) -> int:
             )
         )
 
-    @cached_property
-    def operator_extra_link_dict(self) -> dict[str, Any]:
-        """Returns dictionary of all extra links for the operator."""
-        op_extra_links_from_plugin: dict[str, Any] = {}
-        from airflow import plugins_manager
-
-        plugins_manager.initialize_extra_operators_links_plugins()
-        if plugins_manager.operator_extra_links is None:
-            raise AirflowException("Can't load operators")
-        for ope in plugins_manager.operator_extra_links:
-            if ope.operators and self.operator_class in ope.operators:
-                op_extra_links_from_plugin.update({ope.name: ope})
-
-        operator_extra_links_all = {link.name: link for link in self.operator_extra_links}
-        # Extra links defined in Plugins overrides operator links defined in operator
-        operator_extra_links_all.update(op_extra_links_from_plugin)
-
-        return operator_extra_links_all
-
-    @cached_property
-    def global_operator_extra_link_dict(self) -> dict[str, Any]:
-        """Returns dictionary of all global extra links."""
-        from airflow import plugins_manager
-
-        plugins_manager.initialize_extra_operators_links_plugins()
-        if plugins_manager.global_operator_extra_links is None:
-            raise AirflowException("Can't load operators")
-        return {link.name: link for link in plugins_manager.global_operator_extra_links}
-
-    @cached_property
-    def extra_links(self) -> list[str]:
-        return sorted(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
-
-    def get_extra_links(self, ti: TaskInstance, link_name: str) -> str | None:
-        """
-        For an operator, gets the URLs that the ``extra_links`` entry points to.
-
-        :meta private:
-
-        :raise ValueError: The error message of a ValueError will be passed on through to
-            the fronted to show up as a tooltip on the disabled link.
-        :param ti: The TaskInstance for the URL being searched for.
-        :param link_name: The name of the link we're looking for the URL for. Should be
-            one of the options specified in ``extra_links``.
-        """
-        link: BaseOperatorLink | None = self.operator_extra_link_dict.get(link_name)
-        if not link:
-            link = self.global_operator_extra_link_dict.get(link_name)
-            if not link:
-                return None
-
-        parameters = inspect.signature(link.get_link).parameters
-        old_signature = all(name != "ti_key" for name, p in parameters.items() if p.kind != p.VAR_KEYWORD)
-
-        if old_signature:
-            return link.get_link(self.unmap(None), ti.dag_run.logical_date)  # type: ignore[misc]
-        return link.get_link(self.unmap(None), ti_key=ti.key)
-
     def expand_mapped_task(self, run_id: str, *, session: Session) -> tuple[Sequence[TaskInstance], int]:
         """
         Create the mapped task instances for mapped task.
diff --git a/airflow/models/baseoperatorlink.py b/airflow/models/baseoperatorlink.py
index 952d21bb5592d..6601ee47de923 100644
--- a/airflow/models/baseoperatorlink.py
+++ b/airflow/models/baseoperatorlink.py
@@ -20,14 +20,47 @@
 from abc import ABCMeta, abstractmethod
 from typing import TYPE_CHECKING, ClassVar
 
-import attr
+import attrs
+
+from airflow.models.xcom import BaseXCom
+from airflow.utils.log.logging_mixin import LoggingMixin
 
 if TYPE_CHECKING:
     from airflow.models.baseoperator import BaseOperator
     from airflow.models.taskinstancekey import TaskInstanceKey
 
 
-@attr.s(auto_attribs=True)
+@attrs.define()
+class GenericOperatorLink(LoggingMixin):
+    """A generic operator link class that can retrieve link only using XCOMs. Used while deserializing operators."""
+
+    name: str
+    xcom_key: str
+
+    def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) -> str:
+        """
+        Retrieve the link from the XComs.
+
+        :param operator: The Airflow operator object this link is associated to.
+        :param ti_key: TaskInstance ID to return link for.
+        :return: link to external system, but by pulling it from XComs
+        """
+        self.log.info(
+            "Attempting to retrieve link from XComs with key: %s for task id: %s", self.xcom_key, ti_key
+        )
+        value = BaseXCom.get_one(key=self.xcom_key, run_id=ti_key.run_id)
+        if not value:
+            self.log.debug(
+                "No link with name: %s present in XCom as key: %s, returning empty link",
+                self.name,
+                self.xcom_key,
+            )
+            return ""
+        # Stripping is a temporary workaround till https://github.com/apache/airflow/issues/46513 is handled.
+        return value.strip('"')
+
+
+@attrs.define()
 class BaseOperatorLink(metaclass=ABCMeta):
     """Abstract base class that defines how we get an operator link."""
 
@@ -44,6 +77,17 @@ class BaseOperatorLink(metaclass=ABCMeta):
     def name(self) -> str:
         """Name of the link. This will be the button name on the task UI."""
 
+    @property
+    def xcom_key(self) -> str:
+        """
+        XCom key with while the whole "link" for this operator link is stored.
+
+        On retrieving with this key, the entire link is returned.
+
+        Defaults to `_link_<class name>` if not provided.
+        """
+        return f"_link_{self.__class__.__name__}"
+
     @abstractmethod
     def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) -> str:
         """
diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py
index 08d032e873c6a..7959bdc9d06cc 100644
--- a/airflow/serialization/serialized_objects.py
+++ b/airflow/serialization/serialized_objects.py
@@ -40,13 +40,14 @@
 from airflow.callbacks.callback_requests import DagCallbackRequest, TaskCallbackRequest
 from airflow.exceptions import AirflowException, SerializationError, TaskDeferred
 from airflow.models.baseoperator import BaseOperator
+from airflow.models.baseoperatorlink import BaseOperatorLink, GenericOperatorLink
 from airflow.models.connection import Connection
 from airflow.models.dag import DAG, _get_model_data_interval
 from airflow.models.expandinput import (
     EXPAND_INPUT_EMPTY,
     create_expand_input,
 )
-from airflow.models.taskinstance import SimpleTaskInstance
+from airflow.models.taskinstance import SimpleTaskInstance, TaskInstance
 from airflow.models.taskinstancekey import TaskInstanceKey
 from airflow.models.xcom_arg import SchedulerXComArg, deserialize_xcom_arg
 from airflow.providers_manager import ProvidersManager
@@ -96,7 +97,6 @@
     from inspect import Parameter
 
     from airflow.models import DagRun
-    from airflow.models.baseoperatorlink import BaseOperatorLink
     from airflow.models.expandinput import ExpandInput
     from airflow.sdk.definitions._internal.node import DAGNode
     from airflow.sdk.types import Operator
@@ -1166,6 +1166,25 @@ def __init__(self, *args, **kwargs):
         self.template_fields = BaseOperator.template_fields
         self.operator_extra_links = BaseOperator.operator_extra_links
 
+    def get_extra_links(self, ti: TaskInstance, link_name: str) -> str | None:
+        """
+        For an operator, gets the URLs that the ``extra_links`` entry points to.
+
+        :meta private:
+
+        :raise ValueError: The error message of a ValueError will be passed on through to
+            the fronted to show up as a tooltip on the disabled link.
+        :param ti: The TaskInstance for the URL being searched for.
+        :param link_name: The name of the link we're looking for the URL for. Should be
+            one of the options specified in ``extra_links``.
+        """
+        link = self.operator_extra_link_dict.get(link_name)
+        if not link:
+            link = self.global_operator_extra_link_dict.get(link_name)
+            if not link:
+                return None
+        return link.get_link(self.unmap(None), ti_key=ti.key)
+
     @property
     def task_type(self) -> str:
         # Overwrites task_type of BaseOperator to use _task_type instead of
@@ -1307,7 +1326,8 @@ def populate_operator(cls, op: Operator, encoded_op: dict[str, Any]) -> None:
                         operator.__name__ == encoded_op["task_type"]
                         and operator.__module__ == encoded_op["_task_module"]
                     ):
-                        op_extra_links_from_plugin.update({ope.name: ope})
+                        generic_ope = GenericOperatorLink(name=ope.name, xcom_key=ope.xcom_key)
+                        op_extra_links_from_plugin.update({ope.name: generic_ope})
 
             # If OperatorLinks are defined in Plugins but not in the Operator that is being Serialized
             # set the Operator links attribute
@@ -1503,7 +1523,9 @@ def _is_excluded(cls, var: Any, attrname: str, op: DAGNode):
         return super()._is_excluded(var, attrname, op)
 
     @classmethod
-    def _deserialize_operator_extra_links(cls, encoded_op_links: list) -> dict[str, BaseOperatorLink]:
+    def _deserialize_operator_extra_links(
+        cls, encoded_op_links: dict[str, str]
+    ) -> dict[str, GenericOperatorLink]:
         """
         Deserialize Operator Links if the Classes are registered in Airflow Plugins.
 
@@ -1520,77 +1542,40 @@ def _deserialize_operator_extra_links(cls, encoded_op_links: list) -> dict[str,
             raise AirflowException("Can't load plugins")
         op_predefined_extra_links = {}
 
-        for _operator_links_source in encoded_op_links:
-            # Get the key, value pair as Tuple where key is OperatorLink ClassName
-            # and value is the dictionary containing the arguments passed to the OperatorLink
-            #
-            # Example of a single iteration:
-            #
-            #   _operator_links_source =
-            #   {
-            #       'airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleIndexableLink': {
-            #           'index': 0
-            #       }
-            #   },
+        for name, xcom_key in encoded_op_links.items():
+            # Get the name and xcom_key of the encoded operator and use it to create a GenericOperatorLink object
+            # during deserialization.
             #
-            #   list(_operator_links_source.items()) =
-            #   [
-            #       (
-            #           'airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleIndexableLink',
-            #           {'index': 0}
-            #       )
-            #   ]
-            #
-            #   list(_operator_links_source.items())[0] =
-            #   (
-            #       'airflow.providers.google.cloud.operators.bigquery.BigQueryConsoleIndexableLink',
-            #       {
-            #           'index': 0
-            #       }
-            #   )
-
-            _operator_link_class_path, data = next(iter(_operator_links_source.items()))
-            if _operator_link_class_path in get_operator_extra_links():
-                single_op_link_class = import_string(_operator_link_class_path)
-            elif _operator_link_class_path in plugins_manager.registered_operator_link_classes:
-                single_op_link_class = plugins_manager.registered_operator_link_classes[
-                    _operator_link_class_path
-                ]
-            else:
-                log.error("Operator Link class %r not registered", _operator_link_class_path)
-                return {}
-
-            op_link_parameters = {param: cls.deserialize(value) for param, value in data.items()}
-            op_predefined_extra_link: BaseOperatorLink = single_op_link_class(**op_link_parameters)
-
+            # Example:
+            # enc_operator['_operator_extra_links'] =
+            # {
+            #     'airflow': 'airflow_link_key',
+            #     'foo-bar': 'link-key',
+            #     'no_response': 'key',
+            #     'raise_error': 'key'
+            # }
+
+            op_predefined_extra_link = GenericOperatorLink(name=name, xcom_key=xcom_key)
             op_predefined_extra_links.update({op_predefined_extra_link.name: op_predefined_extra_link})
 
         return op_predefined_extra_links
 
     @classmethod
-    def _serialize_operator_extra_links(cls, operator_extra_links: Iterable[BaseOperatorLink]):
+    def _serialize_operator_extra_links(
+        cls, operator_extra_links: Iterable[BaseOperatorLink]
+    ) -> dict[str, str]:
         """
         Serialize Operator Links.
 
-        Store the import path of the OperatorLink and the arguments passed to it.
+        Store the "name" of the link mapped with the xcom_key which can be later used to retrieve this
+        operator extra link from XComs.
         For example:
-        ``[{'airflow.providers.google.cloud.links.bigquery.BigQueryDatasetLink': {}}]``
+        ``{'link-name-1': 'xcom-key-1'}``
 
         :param operator_extra_links: Operator Link
         :return: Serialized Operator Link
         """
-        serialize_operator_extra_links = []
-        for operator_extra_link in operator_extra_links:
-            op_link_arguments = {
-                param: cls.serialize(value) for param, value in attrs.asdict(operator_extra_link).items()
-            }
-
-            module_path = (
-                f"{operator_extra_link.__class__.__module__}.{operator_extra_link.__class__.__name__}"
-            )
-            serialize_operator_extra_links.append({module_path: op_link_arguments})
-
-        return serialize_operator_extra_links
+        return {link.name: link.xcom_key for link in operator_extra_links}
 
     @classmethod
     def serialize(cls, var: Any, *, strict: bool = False) -> Any:
diff --git a/newsfragments/46613.feature.rst b/newsfragments/46613.feature.rst
new file mode 100644
index 0000000000000..3f84bb48d8e46
--- /dev/null
+++ b/newsfragments/46613.feature.rst
@@ -0,0 +1 @@
+Operator Links interface changed to not run user code in Airflow Webserver The Operator Extra links, which can be defined either via plugins or custom operators now do not execute any user code in the Airflow Webserver, but instead push the "full" links to XCom backend and the value is again fetched from the XCom backend when viewing task details in grid view.
diff --git a/task_sdk/src/airflow/sdk/definitions/_internal/abstractoperator.py b/task_sdk/src/airflow/sdk/definitions/_internal/abstractoperator.py
index 481fb2ee773d7..281ebd04b4543 100644
--- a/task_sdk/src/airflow/sdk/definitions/_internal/abstractoperator.py
+++ b/task_sdk/src/airflow/sdk/definitions/_internal/abstractoperator.py
@@ -27,10 +27,12 @@
     Iterator,
     Mapping,
 )
+from functools import cached_property
 from typing import TYPE_CHECKING, Any, ClassVar
 
 import methodtools
 
+from airflow.exceptions import AirflowException
 from airflow.sdk.definitions._internal.mixins import DependencyMixin
 from airflow.sdk.definitions._internal.node import DAGNode
 from airflow.sdk.definitions._internal.templater import Templater
@@ -136,6 +138,39 @@ class AbstractOperator(Templater, DAGNode):
         )
     )
 
+    @cached_property
+    def operator_extra_link_dict(self) -> dict[str, Any]:
+        """Returns dictionary of all extra links for the operator."""
+        op_extra_links_from_plugin: dict[str, Any] = {}
+        from airflow import plugins_manager
+
+        plugins_manager.initialize_extra_operators_links_plugins()
+        if plugins_manager.operator_extra_links is None:
+            raise AirflowException("Can't load operators")
+        for ope in plugins_manager.operator_extra_links:
+            if ope.operators and self.operator_class in ope.operators:
+                op_extra_links_from_plugin.update({ope.name: ope})
+
+        operator_extra_links_all = {link.name: link for link in self.operator_extra_links}
+        # Extra links defined in Plugins overrides operator links defined in operator
+        operator_extra_links_all.update(op_extra_links_from_plugin)
+
+        return operator_extra_links_all
+
+    @cached_property
+    def global_operator_extra_link_dict(self) -> dict[str, Any]:
+        """Returns dictionary of all global extra links."""
+        from airflow import plugins_manager
+
+        plugins_manager.initialize_extra_operators_links_plugins()
+        if plugins_manager.global_operator_extra_links is None:
+            raise AirflowException("Can't load operators")
+        return {link.name: link for link in plugins_manager.global_operator_extra_links}
+
+    @cached_property
+    def extra_links(self) -> list[str]:
+        return sorted(set(self.operator_extra_link_dict).union(self.global_operator_extra_link_dict))
+
     def get_dag(self) -> DAG | None:
         raise NotImplementedError()
 
diff --git a/task_sdk/src/airflow/sdk/execution_time/task_runner.py b/task_sdk/src/airflow/sdk/execution_time/task_runner.py
index 77cd29d090c2d..c34d2d79e4434 100644
--- a/task_sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/task_sdk/src/airflow/sdk/execution_time/task_runner.py
@@ -717,7 +717,11 @@ def _push_xcom_if_needed(result: Any, ti: RuntimeTaskInstance):
     _xcom_push(ti, "return_value", result, mapped_length=mapped_length)
 
 
-def finalize(log: Logger): ...
+def finalize(ti: RuntimeTaskInstance, log: Logger):
+    for oe in ti.task.operator_extra_link_dict.values():
+        link, xcom_key = oe.get_link(operator=ti.task, ti_key=ti.id), oe.xcom_key  # type: ignore[arg-type]
+        log.debug("Setting xcom for operator extra link", link=link, xcom_key=xcom_key)
+        _xcom_push(ti, key=xcom_key, value=link)
 
 
 def main():
@@ -728,7 +732,7 @@ def main():
     try:
         ti, log = startup()
         run(ti, log)
-        finalize(log)
+        finalize(ti, log)
     except KeyboardInterrupt:
         log = structlog.get_logger(logger_name="task")
         log.exception("Ctrl-c hit")
diff --git a/task_sdk/tests/execution_time/test_task_runner.py b/task_sdk/tests/execution_time/test_task_runner.py
index 85804836981d0..bde77e95cd3b4 100644
--- a/task_sdk/tests/execution_time/test_task_runner.py
+++ b/task_sdk/tests/execution_time/test_task_runner.py
@@ -55,6 +55,7 @@
     PrevSuccessfulDagRunResult,
     RuntimeCheckOnTask,
     SetRenderedFields,
+    SetXCom,
     StartupDetails,
     SucceedTask,
     TaskState,
@@ -72,12 +73,15 @@
     RuntimeTaskInstance,
     _push_xcom_if_needed,
     _xcom_push,
+    finalize,
     parse,
     run,
     startup,
 )
 from airflow.utils import timezone
 
+from tests_common.test_utils.mock_operators import AirflowLink
+
 FAKE_BUNDLE = BundleInfo(name="anything", version="any")
 
 
@@ -1107,6 +1111,46 @@ def execute(self, context):
         msg = run(runtime_ti, log=mock.MagicMock())
         assert isinstance(msg, SucceedTask)
 
+    def test_task_run_with_operator_extra_links(self, create_runtime_ti, mock_supervisor_comms, time_machine):
+        """Test that a task can run with operator extra links defined and can set an xcom."""
+        instant = timezone.datetime(2024, 12, 3, 10, 0)
+        time_machine.move_to(instant, tick=False)
+
+        class DummyTestOperator(BaseOperator):
+            operator_extra_links = (AirflowLink(),)
+
+            def execute(self, context):
+                print("Hello from custom operator", self.operator_extra_links)
+
+        task = DummyTestOperator(task_id="task_with_operator_extra_links")
+
+        runtime_ti = create_runtime_ti(task=task)
+
+        run(runtime_ti, log=mock.MagicMock())
+
+        mock_supervisor_comms.send_request.assert_called_once_with(
+            msg=SucceedTask(
+                state=TerminalTIState.SUCCESS, end_date=instant, task_outlets=[], outlet_events=[]
+            ),
+            log=mock.ANY,
+        )
+
+        finalize(runtime_ti, log=mock.MagicMock())
+
+        mock_supervisor_comms.send_request.assert_any_call(
+            msg=SetXCom(
+                key="_link_AirflowLink",
+                value='"https://airflow.apache.org"',
+                dag_id="test_dag",
+                run_id="test_run",
+                task_id="task_with_operator_extra_links",
+                map_index=-1,
+                mapped_length=None,
+                type="SetXCom",
+            ),
+            log=mock.ANY,
+        )
+
 
 class TestXComAfterTaskExecution:
     @pytest.mark.parametrize(


Code review comments:
diff_hunk: @@ -1166,6 +1166,25 @@ def __init__(self, *args, **kwargs):
         self.template_fields = BaseOperator.template_fields
         self.operator_extra_links = BaseOperator.operator_extra_links
 
+    def get_extra_links(self, ti: TaskInstance, link_name: str) -> str | None:
+        """
+        For an operator, gets the URLs that the ``extra_links`` entry points to.
+
+        :meta private:
+
+        :raise ValueError: The error message of a ValueError will be passed on through to
+            the fronted to show up as a tooltip on the disabled link.
+        :param ti: The TaskInstance for the URL being searched for.
+        :param link_name: The name of the link we're looking for the URL for. Should be
+            one of the options specified in ``extra_links``.
+        """
+        link = self.operator_extra_link_dict.get(link_name)
+        if not link:
+            link = self.global_operator_extra_link_dict.get(link_name)
+            if not link:
+                return None
+        return link.get_link(self.unmap(None), ti_key=ti.key)
path: airflow/serialization/serialized_objects.py
text: @author:
Can this stay this way or does it have to stay like the previous definition:
```
        link: BaseOperatorLink | None = self.operator_extra_link_dict.get(link_name)
        if not link:
            link = self.global_operator_extra_link_dict.get(link_name)
            if not link:
                return None

        parameters = inspect.signature(link.get_link).parameters
        old_signature = all(name != "ti_key" for name, p in parameters.items() if p.kind != p.VAR_KEYWORD)

        if old_signature:
            return link.get_link(self.unmap(None), ti.dag_run.logical_date)  # type: ignore[misc]
        return link.get_link(self.unmap(None), ti_key=ti.key)
```

@user1:
See #46415

@author:
Thanks a lot @user1!! I was going to propose we remove the old signature anyways, thats helpful

@author:
Resolving based on your comment, irrelevant now