You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by hu...@apache.org on 2024/01/09 23:29:31 UTC
(airflow) branch main updated: Support encryption for triggers parameters (#36492)
This is an automated email from the ASF dual-hosted git repository.
husseinawala pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 8fb55f2127 Support encryption for triggers parameters (#36492)
8fb55f2127 is described below
commit 8fb55f2127be379bf25ed5fd6d2933417e8da439
Author: Hussein Awala <hu...@awala.fr>
AuthorDate: Wed Jan 10 00:29:24 2024 +0100
Support encryption for triggers parameters (#36492)
* Implement a way to encrypt some trigger params stored in the DB
* Update deferring doc
* Create a variable for encryption prefix
* Update the test
---
airflow/jobs/triggerer_job_runner.py | 27 +++++++++---
airflow/models/trigger.py | 13 +++++-
.../authoring-and-scheduling/deferring.rst | 27 ++++++++++++
tests/models/test_trigger.py | 49 +++++++++++++++++++++-
4 files changed, 108 insertions(+), 8 deletions(-)
diff --git a/airflow/jobs/triggerer_job_runner.py b/airflow/jobs/triggerer_job_runner.py
index bb151b32cc..b25ce6a7d3 100644
--- a/airflow/jobs/triggerer_job_runner.py
+++ b/airflow/jobs/triggerer_job_runner.py
@@ -28,16 +28,16 @@ from collections import deque
from contextlib import suppress
from copy import copy
from queue import SimpleQueue
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, TypeVar
from sqlalchemy import func, select
from airflow.configuration import conf
from airflow.jobs.base_job_runner import BaseJobRunner
from airflow.jobs.job import perform_heartbeat
-from airflow.models.trigger import Trigger
+from airflow.models.trigger import ENCRYPTED_KWARGS_PREFIX, Trigger
from airflow.stats import Stats
-from airflow.triggers.base import TriggerEvent
+from airflow.triggers.base import BaseTrigger, TriggerEvent
from airflow.typing_compat import TypedDict
from airflow.utils import timezone
from airflow.utils.log.file_task_handler import FileTaskHandler
@@ -60,7 +60,6 @@ if TYPE_CHECKING:
from airflow.jobs.job import Job
from airflow.models import TaskInstance
- from airflow.triggers.base import BaseTrigger
HANDLER_SUPPORTS_TRIGGERER = False
"""
@@ -236,6 +235,9 @@ def setup_queue_listener():
return None
+U = TypeVar("U", bound=BaseTrigger)
+
+
class TriggererJobRunner(BaseJobRunner, LoggingMixin):
"""
Run active triggers in asyncio and update their dependent tests/DAGs once their events have fired.
@@ -673,7 +675,7 @@ class TriggerRunner(threading.Thread, LoggingMixin):
continue
try:
- new_trigger_instance = trigger_class(**new_trigger_orm.kwargs)
+ new_trigger_instance = self.trigger_row_to_trigger_instance(new_trigger_orm, trigger_class)
except TypeError as err:
self.log.error("Trigger failed; message=%s", err)
self.failed_triggers.append((new_id, err))
@@ -708,3 +710,18 @@ class TriggerRunner(threading.Thread, LoggingMixin):
if classpath not in self.trigger_cache:
self.trigger_cache[classpath] = import_string(classpath)
return self.trigger_cache[classpath]
+
+ def trigger_row_to_trigger_instance(self, trigger_row: Trigger, trigger_class: type[U]) -> U:
+ """Convert a Trigger row into a Trigger instance."""
+ from airflow.models.crypto import get_fernet
+
+ decrypted_kwargs = {}
+ fernet = get_fernet()
+ for k, v in trigger_row.kwargs.items():
+ if k.startswith(ENCRYPTED_KWARGS_PREFIX):
+ decrypted_kwargs[k[len(ENCRYPTED_KWARGS_PREFIX) :]] = fernet.decrypt(
+ v.encode("utf-8")
+ ).decode("utf-8")
+ else:
+ decrypted_kwargs[k] = v
+ return trigger_class(**decrypted_kwargs)
diff --git a/airflow/models/trigger.py b/airflow/models/trigger.py
index 4ad42b17b8..2a00ab3856 100644
--- a/airflow/models/trigger.py
+++ b/airflow/models/trigger.py
@@ -38,6 +38,8 @@ if TYPE_CHECKING:
from airflow.triggers.base import BaseTrigger
+ENCRYPTED_KWARGS_PREFIX = "encrypted__"
+
class Trigger(Base):
"""
@@ -90,8 +92,17 @@ class Trigger(Base):
@internal_api_call
def from_object(cls, trigger: BaseTrigger) -> Trigger:
"""Alternative constructor that creates a trigger row based directly off of a Trigger object."""
+ from airflow.models.crypto import get_fernet
+
classpath, kwargs = trigger.serialize()
- return cls(classpath=classpath, kwargs=kwargs)
+ secure_kwargs = {}
+ fernet = get_fernet()
+ for k, v in kwargs.items():
+ if k.startswith(ENCRYPTED_KWARGS_PREFIX):
+ secure_kwargs[k] = fernet.encrypt(v.encode("utf-8")).decode("utf-8")
+ else:
+ secure_kwargs[k] = v
+ return cls(classpath=classpath, kwargs=secure_kwargs)
@classmethod
@internal_api_call
diff --git a/docs/apache-airflow/authoring-and-scheduling/deferring.rst b/docs/apache-airflow/authoring-and-scheduling/deferring.rst
index adacf4b79d..0702793e2b 100644
--- a/docs/apache-airflow/authoring-and-scheduling/deferring.rst
+++ b/docs/apache-airflow/authoring-and-scheduling/deferring.rst
@@ -197,6 +197,33 @@ Triggers can be as complex or as simple as you want, provided they meet the desi
If you are new to writing asynchronous Python, be very careful when writing your ``run()`` method. Python's async model means that code can block the entire process if it does not correctly ``await`` when it does a blocking operation. Airflow attempts to detect process blocking code and warn you in the triggerer logs when it happens. You can enable extra checks by Python by setting the variable ``PYTHONASYNCIODEBUG=1`` when you are writing your trigger to make sure you're writing non-blo [...]
+Sensitive information in triggers
+'''''''''''''''''''''''''''''''''
+
+Triggers are serialized and stored in the database, so they can be re-instantiated on any triggerer process. This means that any sensitive information you pass to a trigger will be stored in the database.
+If you want to pass sensitive information to a trigger, you can encrypt it before passing it to the trigger, and decrypt it inside the trigger, or update the argument name in the ``serialize`` method by adding ``encrypted__`` as a prefix, and Airflow will automatically encrypt the argument before storing it in the database, and decrypt it when it is read from the database.
+
+.. code-block:: python
+
+ class MyTrigger(BaseTrigger):
+ def __init__(self, param, secret):
+ super().__init__()
+ self.param = param
+ self.secret = secret
+
+ def serialize(self):
+ return (
+ "airflow.triggers.MyTrigger",
+ {
+ "param": self.param,
+ "encrypted__secret": self.secret,
+ },
+ )
+
+ async def run(self):
+ # self.my_secret will be decrypted here
+ ...
+
High Availability
-----------------
diff --git a/tests/models/test_trigger.py b/tests/models/test_trigger.py
index 8463edf831..ce1858443e 100644
--- a/tests/models/test_trigger.py
+++ b/tests/models/test_trigger.py
@@ -17,18 +17,21 @@
from __future__ import annotations
import datetime
+from typing import Any, AsyncIterator
import pytest
import pytz
+from cryptography.fernet import Fernet
from airflow.jobs.job import Job
-from airflow.jobs.triggerer_job_runner import TriggererJobRunner
+from airflow.jobs.triggerer_job_runner import TriggererJobRunner, TriggerRunner
from airflow.models import TaskInstance, Trigger
from airflow.operators.empty import EmptyOperator
-from airflow.triggers.base import TriggerEvent
+from airflow.triggers.base import BaseTrigger, TriggerEvent
from airflow.utils import timezone
from airflow.utils.session import create_session
from airflow.utils.state import State
+from tests.test_utils.config import conf_vars
pytestmark = pytest.mark.db_test
@@ -337,3 +340,45 @@ def test_get_sorted_triggers_different_priority_weights(session, create_task_ins
trigger_ids_query = Trigger.get_sorted_triggers(capacity=100, alive_triggerer_ids=[], session=session)
assert trigger_ids_query == [(2,), (1,)]
+
+
+class SensitiveKwargsTrigger(BaseTrigger):
+ """
+ A trigger that has sensitive kwargs.
+ """
+
+ def __init__(self, param1: str, param2: str):
+ super().__init__()
+ self.param1 = param1
+ self.param2 = param2
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ return (
+ "tests.models.test_trigger.SensitiveKwargsTrigger",
+ {
+ "param1": self.param1,
+ "encrypted__param2": self.param2,
+ },
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ yield TriggerEvent({})
+
+
+@conf_vars({("core", "fernet_key"): Fernet.generate_key().decode()})
+def test_serialize_sensitive_kwargs():
+ """
+ Tests that sensitive kwargs are encrypted.
+ """
+ trigger_instance = SensitiveKwargsTrigger(param1="value1", param2="value2")
+ trigger_row: Trigger = Trigger.from_object(trigger_instance)
+
+ assert trigger_row.kwargs["param1"] == "value1"
+ assert "param2" not in trigger_row.kwargs
+ assert trigger_row.kwargs["encrypted__param2"] != "value2"
+
+ loaded_trigger: SensitiveKwargsTrigger = TriggerRunner().trigger_row_to_trigger_instance(
+ trigger_row, SensitiveKwargsTrigger
+ )
+ assert loaded_trigger.param1 == "value1"
+ assert loaded_trigger.param2 == "value2"