Add a migration script for encrypted trigger kwargs

airflow/migrations/versions/0138_2_9_0_encrypt_trigger_kwargs.py

@@ -0,0 +1,69 @@
+#
+# 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.
+
+"""encrypt trigger kwargs
+
+Revision ID: 1949afb29106
+Revises: 8e1c784a4fc7
+Create Date: 2024-03-17 22:09:09.406395
+
+"""
+import sqlalchemy as sa
+from airflow.models.crypto import get_fernet
+
+from airflow.models.trigger import Trigger
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = "1949afb29106"
+down_revision = "8e1c784a4fc7"
+branch_labels = None
+depends_on = None
+airflow_version = "2.9.0"
+
+
+def get_session() -> sa.orm.Session:
+    conn = op.get_bind()
+    sessionmaker = sa.orm.sessionmaker()
+    return sessionmaker(bind=conn)
+
+
+def upgrade():
+    """Apply encrypt trigger kwargs"""
+    session = get_session()
+    try:
+        for trigger in session.query(Trigger).all():
+            # convert dict to string and encrypt it
+            trigger.encrypted_kwargs = trigger._encrypt_kwargs(trigger.encrypted_kwargs)
+        session.commit()
+    finally:
+        session.close()
+
+
+def downgrade():
+    """Unapply encrypt trigger kwargs"""
+    session = get_session()
+    try:
+        fernet = get_fernet()
+        for trigger in session.query(Trigger).all():
+            # decrypt string and convert it to dict
+            trigger.encrypted_kwargs = trigger._decrypt_kwargs(trigger.encrypted_kwargs, fernet)
+        session.commit()
+    finally:
+        session.close()

airflow/utils/db.py

@@ -90,7 +90,7 @@
     "2.7.0": "405de8318b3a",
     "2.8.0": "10b52ebd31f7",
     "2.8.1": "88344c1d9134",
-    "2.9.0": "1fd565369930",
+    "2.9.0": "1949afb29106",
 }
 
 

docs/apache-airflow/img/airflow_erd.sha256

@@ -1 +1 @@
-ce71fa2d8f6daea541c46d3528961bdee87421b3612ed2f2451e88630a52fd70
+c6a12adc3097ace8272f6a454b952092b65de1cf5a65fc6760f7fd618098281a

docs/apache-airflow/migrations-ref.rst

@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
 | Revision ID                     | Revises ID        | Airflow Version   | Description                                                  |
 +=================================+===================+===================+==============================================================+
-| ``8e1c784a4fc7`` (head)         | ``ab34f260b71c``  | ``2.9.0``         | Adding max_consecutive_failed_dag_runs column to dag_model   |
+| ``1949afb29106`` (head)         | ``8e1c784a4fc7``  | ``2.9.0``         | encrypt trigger kwargs                                       |
++---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
+| ``8e1c784a4fc7``                | ``ab34f260b71c``  | ``2.9.0``         | Adding max_consecutive_failed_dag_runs column to dag_model   |
 |                                 |                   |                   | table                                                        |
 +---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
 | ``ab34f260b71c``                | ``d75389605139``  | ``2.9.0``         | add dataset_expression in DagModel                           |