You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "jedcunningham (via GitHub)" <gi...@apache.org> on 2024/04/25 05:32:01 UTC
[PR] Fix trigger kwarg encryption migration [airflow]
jedcunningham opened a new pull request, #39246:
URL: https://github.com/apache/airflow/pull/39246
Closes: #38836
Alternative to #38876
Do the encryption in the migration itself, and fix support for offline migrations as well.
The offline up migration won't actually encrypt the trigger kwargs as there isn't a safe way to accomplish that, so the decryption processes checks and short circuits if it isn't encrypted.
The offline down migration will now print out a warning that the offline migration will fail if there are any running triggers. I think this is the best we can do for that scenario (and folks willing to do offline migrations will hopefully be able to understand the situation).
This also solves the "encrypting the already encrypted kwargs" bug in 2.9.0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1579934629
##########
airflow/migrations/versions/0140_2_9_0_update_trigger_kwargs_type.py:
##########
@@ -38,13 +42,43 @@
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():
- """Update trigger kwargs type to string"""
+ """Update trigger kwargs type to string and encrypt"""
with op.batch_alter_table("trigger") as batch_op:
batch_op.alter_column("kwargs", type_=sa.Text(), )
+ if not context.is_offline_mode():
+ session = get_session()
+ try:
+ for trigger in session.query(Trigger).options(lazyload(Trigger.task_instance)):
+ trigger.kwargs = trigger.kwargs
+ session.commit()
+ finally:
+ session.close()
Review Comment:
Yeah, this is something I was a little suspect of also, but haven't had a chance to verify it yet. Let me do that now.
That said, this is a pattern we have in existing migrations already π€·.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham merged PR #39246:
URL: https://github.com/apache/airflow/pull/39246
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1578968682
##########
airflow/migrations/versions/0140_2_9_0_update_trigger_kwargs_type.py:
##########
@@ -38,13 +41,43 @@
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():
- """Update trigger kwargs type to string"""
+ """Update trigger kwargs type to string and encrypt"""
with op.batch_alter_table("trigger") as batch_op:
batch_op.alter_column("kwargs", type_=sa.Text(), )
+ if not context.is_offline_mode():
+ session = get_session()
+ try:
+ for trigger in session.query(Trigger):
+ trigger.kwargs = trigger.kwargs
+ session.commit()
+ finally:
+ session.close()
+
def downgrade():
- """Unapply update trigger kwargs type to string"""
+ """Unapply update trigger kwargs type to string and encrypt"""
+ if context.is_offline_mode():
+ print(dedent("""
+ ------------
+ -- WARNING: Unable to decrypt trigger kwargs automatically in offline mode!
Review Comment:
It's a moving target. There may be none now but when they go run the generated sql it blows up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1579930673
##########
airflow/migrations/versions/0140_2_9_0_update_trigger_kwargs_type.py:
##########
@@ -38,13 +42,43 @@
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():
- """Update trigger kwargs type to string"""
+ """Update trigger kwargs type to string and encrypt"""
with op.batch_alter_table("trigger") as batch_op:
batch_op.alter_column("kwargs", type_=sa.Text(), )
+ if not context.is_offline_mode():
+ session = get_session()
+ try:
+ for trigger in session.query(Trigger).options(lazyload(Trigger.task_instance)):
+ trigger.kwargs = trigger.kwargs
+ session.commit()
+ finally:
+ session.close()
Review Comment:
i'm fuzzy on this... but are all upgrades supposed to be in a single transaction?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1579958752
##########
airflow/migrations/versions/0140_2_9_0_update_trigger_kwargs_type.py:
##########
@@ -38,13 +42,43 @@
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():
- """Update trigger kwargs type to string"""
+ """Update trigger kwargs type to string and encrypt"""
with op.batch_alter_table("trigger") as batch_op:
batch_op.alter_column("kwargs", type_=sa.Text(), )
+ if not context.is_offline_mode():
+ session = get_session()
+ try:
+ for trigger in session.query(Trigger).options(lazyload(Trigger.task_instance)):
+ trigger.kwargs = trigger.kwargs
+ session.commit()
+ finally:
+ session.close()
Review Comment:
Yeah it behaves like a "child session", which is kinda cool.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1578966905
##########
airflow/migrations/versions/0140_2_9_0_update_trigger_kwargs_type.py:
##########
@@ -38,13 +41,43 @@
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():
- """Update trigger kwargs type to string"""
+ """Update trigger kwargs type to string and encrypt"""
with op.batch_alter_table("trigger") as batch_op:
batch_op.alter_column("kwargs", type_=sa.Text(), )
+ if not context.is_offline_mode():
+ session = get_session()
+ try:
+ for trigger in session.query(Trigger):
+ trigger.kwargs = trigger.kwargs
+ session.commit()
+ finally:
+ session.close()
+
def downgrade():
- """Unapply update trigger kwargs type to string"""
+ """Unapply update trigger kwargs type to string and encrypt"""
+ if context.is_offline_mode():
+ print(dedent("""
+ ------------
+ -- WARNING: Unable to decrypt trigger kwargs automatically in offline mode!
Review Comment:
Instead of emitting a warning, would it be possible to detect the case and just fail right here? This would make recovery a bit easier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1578985358
##########
airflow/models/trigger.py:
##########
@@ -116,7 +116,15 @@ def _decrypt_kwargs(encrypted_kwargs: str) -> dict[str, Any]:
from airflow.models.crypto import get_fernet
from airflow.serialization.serialized_objects import BaseSerialization
- decrypted_kwargs = json.loads(get_fernet().decrypt(encrypted_kwargs.encode("utf-8")).decode("utf-8"))
+ # We weren't able to encrypt the kwargs in all migration paths,
+ # so we need to handle the case where they are not encrypted.
+ # Triggers aren't long lasting, so we can skip encrypting them now.
+ if encrypted_kwargs.startswith("{"):
+ decrypted_kwargs = json.loads(encrypted_kwargs)
Review Comment:
Maybe we can catch the exception and run the else block below in that case?
##########
airflow/models/trigger.py:
##########
@@ -116,7 +116,15 @@ def _decrypt_kwargs(encrypted_kwargs: str) -> dict[str, Any]:
from airflow.models.crypto import get_fernet
from airflow.serialization.serialized_objects import BaseSerialization
- decrypted_kwargs = json.loads(get_fernet().decrypt(encrypted_kwargs.encode("utf-8")).decode("utf-8"))
+ # We weren't able to encrypt the kwargs in all migration paths,
+ # so we need to handle the case where they are not encrypted.
+ # Triggers aren't long lasting, so we can skip encrypting them now.
+ if encrypted_kwargs.startswith("{"):
+ decrypted_kwargs = json.loads(encrypted_kwargs)
Review Comment:
I think it's unlikely π€ but is it possible the first letter after encryption is `{`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1579935696
##########
airflow/models/trigger.py:
##########
@@ -116,7 +116,15 @@ def _decrypt_kwargs(encrypted_kwargs: str) -> dict[str, Any]:
from airflow.models.crypto import get_fernet
from airflow.serialization.serialized_objects import BaseSerialization
- decrypted_kwargs = json.loads(get_fernet().decrypt(encrypted_kwargs.encode("utf-8")).decode("utf-8"))
+ # We weren't able to encrypt the kwargs in all migration paths,
+ # so we need to handle the case where they are not encrypted.
+ # Triggers aren't long lasting, so we can skip encrypting them now.
+ if encrypted_kwargs.startswith("{"):
+ decrypted_kwargs = json.loads(encrypted_kwargs)
Review Comment:
FWIW i kindof like EAFP better.... i guess i'm using that acronym now...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1578991552
##########
airflow/models/trigger.py:
##########
@@ -116,7 +116,15 @@ def _decrypt_kwargs(encrypted_kwargs: str) -> dict[str, Any]:
from airflow.models.crypto import get_fernet
from airflow.serialization.serialized_objects import BaseSerialization
- decrypted_kwargs = json.loads(get_fernet().decrypt(encrypted_kwargs.encode("utf-8")).decode("utf-8"))
+ # We weren't able to encrypt the kwargs in all migration paths,
+ # so we need to handle the case where they are not encrypted.
+ # Triggers aren't long lasting, so we can skip encrypting them now.
+ if encrypted_kwargs.startswith("{"):
+ decrypted_kwargs = json.loads(encrypted_kwargs)
Review Comment:
The [result is base64](https://cryptography.io/en/latest/fernet/#cryptography.fernet.Fernet.encrypt), so no risk. But if you want to refactor to decrypt/catch instead, have at it π
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1579040923
##########
airflow/models/trigger.py:
##########
@@ -116,7 +116,15 @@ def _decrypt_kwargs(encrypted_kwargs: str) -> dict[str, Any]:
from airflow.models.crypto import get_fernet
from airflow.serialization.serialized_objects import BaseSerialization
- decrypted_kwargs = json.loads(get_fernet().decrypt(encrypted_kwargs.encode("utf-8")).decode("utf-8"))
+ # We weren't able to encrypt the kwargs in all migration paths,
+ # so we need to handle the case where they are not encrypted.
+ # Triggers aren't long lasting, so we can skip encrypting them now.
Review Comment:
Letβs add an issue to track this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1579944620
##########
airflow/migrations/versions/0140_2_9_0_update_trigger_kwargs_type.py:
##########
@@ -38,13 +42,43 @@
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():
- """Update trigger kwargs type to string"""
+ """Update trigger kwargs type to string and encrypt"""
with op.batch_alter_table("trigger") as batch_op:
batch_op.alter_column("kwargs", type_=sa.Text(), )
+ if not context.is_offline_mode():
+ session = get_session()
+ try:
+ for trigger in session.query(Trigger).options(lazyload(Trigger.task_instance)):
+ trigger.kwargs = trigger.kwargs
+ session.commit()
+ finally:
+ session.close()
Review Comment:
So turns out this doesn't actually commit the DDL OR DML, the "real" commit happens when alembic does it in a higher scope. Should be good here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1579932574
##########
airflow/models/trigger.py:
##########
@@ -116,7 +116,15 @@ def _decrypt_kwargs(encrypted_kwargs: str) -> dict[str, Any]:
from airflow.models.crypto import get_fernet
from airflow.serialization.serialized_objects import BaseSerialization
- decrypted_kwargs = json.loads(get_fernet().decrypt(encrypted_kwargs.encode("utf-8")).decode("utf-8"))
+ # We weren't able to encrypt the kwargs in all migration paths,
+ # so we need to handle the case where they are not encrypted.
+ # Triggers aren't long lasting, so we can skip encrypting them now.
Review Comment:
Done: #39265
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1579948255
##########
airflow/migrations/versions/0140_2_9_0_update_trigger_kwargs_type.py:
##########
@@ -38,13 +42,43 @@
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():
- """Update trigger kwargs type to string"""
+ """Update trigger kwargs type to string and encrypt"""
with op.batch_alter_table("trigger") as batch_op:
batch_op.alter_column("kwargs", type_=sa.Text(), )
+ if not context.is_offline_mode():
+ session = get_session()
+ try:
+ for trigger in session.query(Trigger).options(lazyload(Trigger.task_instance)):
+ trigger.kwargs = trigger.kwargs
+ session.commit()
+ finally:
+ session.close()
Review Comment:
weird. i guess it's a different session.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on PR #39246:
URL: https://github.com/apache/airflow/pull/39246#issuecomment-2077869872
We've tested the matrix of postges/mysql/sqlite and online/offline migrations. Didn't find any issues π.
I think this should be good. I'd like to get 1 more approval on this before we merge it though.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1578993768
##########
airflow/models/trigger.py:
##########
@@ -116,7 +116,15 @@ def _decrypt_kwargs(encrypted_kwargs: str) -> dict[str, Any]:
from airflow.models.crypto import get_fernet
from airflow.serialization.serialized_objects import BaseSerialization
- decrypted_kwargs = json.loads(get_fernet().decrypt(encrypted_kwargs.encode("utf-8")).decode("utf-8"))
+ # We weren't able to encrypt the kwargs in all migration paths,
+ # so we need to handle the case where they are not encrypted.
+ # Triggers aren't long lasting, so we can skip encrypting them now.
+ if encrypted_kwargs.startswith("{"):
+ decrypted_kwargs = json.loads(encrypted_kwargs)
Review Comment:
If there's no risk, I think it's better keep it as it is for now to avoid unnecessary logic π
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1579933342
##########
airflow/migrations/versions/0140_2_9_0_update_trigger_kwargs_type.py:
##########
@@ -38,13 +42,43 @@
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():
- """Update trigger kwargs type to string"""
+ """Update trigger kwargs type to string and encrypt"""
with op.batch_alter_table("trigger") as batch_op:
batch_op.alter_column("kwargs", type_=sa.Text(), )
+ if not context.is_offline_mode():
Review Comment:
very much a style thing, but i think doing a pre-emptive return better reveals the intention in this case
i.e. `if context.is_offline_mode(): return` but feel free to disregard
sorta just makes it easier to see what "the main migration" is ... indented makes it look like it's conditional
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1578894900
##########
airflow/migrations/versions/0140_2_9_0_update_trigger_kwargs_type.py:
##########
@@ -38,13 +41,43 @@
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():
- """Update trigger kwargs type to string"""
+ """Update trigger kwargs type to string and encrypt"""
with op.batch_alter_table("trigger") as batch_op:
batch_op.alter_column("kwargs", type_=sa.Text(), )
+ if not context.is_offline_mode():
+ session = get_session()
+ try:
+ for trigger in session.query(Trigger):
+ trigger.kwargs = trigger.kwargs
+ session.commit()
+ finally:
+ session.close()
+
def downgrade():
- """Unapply update trigger kwargs type to string"""
+ """Unapply update trigger kwargs type to string and encrypt"""
+ if context.is_offline_mode():
+ print(dedent("""
+ ------------
+ -- WARNING: Unable to decrypt trigger kwargs automatically in offline mode!
Review Comment:
I don't think we can do any better than spitting out a warning :(.
But if folks have ideas here, happy to hear them.
##########
airflow/models/trigger.py:
##########
@@ -116,7 +116,15 @@ def _decrypt_kwargs(encrypted_kwargs: str) -> dict[str, Any]:
from airflow.models.crypto import get_fernet
from airflow.serialization.serialized_objects import BaseSerialization
- decrypted_kwargs = json.loads(get_fernet().decrypt(encrypted_kwargs.encode("utf-8")).decode("utf-8"))
+ # We weren't able to encrypt the kwargs in all migration paths,
+ # so we need to handle the case where they are not encrypted.
+ # Triggers aren't long lasting, so we can skip encrypting them now.
Review Comment:
We could start doing this if we'd like to, but I'd suggest we do it in a follow up PR.
##########
tests/models/test_trigger.py:
##########
@@ -378,3 +380,19 @@ def test_serialize_sensitive_kwargs():
assert isinstance(trigger_row.encrypted_kwargs, str)
assert "value1" not in trigger_row.encrypted_kwargs
assert "value2" not in trigger_row.encrypted_kwargs
+
+
+def test_kwargs_not_encrypted():
+ """
+ Tests that we don't decrypt kwargs if they aren't encrypted.
+ We weren't able to encrypt the kwargs in all migration paths.
+ """
+ trigger = Trigger(classpath="airflow.triggers.testing.SuccessTrigger", kwargs={})
+ # force the `encrypted_kwargs` to be unencrypted, like they would be after an upgrade
+ trigger.encrypted_kwargs = json.dumps(
+ BaseSerialization.serialize({"param1": "value1", "param2": "value2"})
+ )
+ print(trigger.encrypted_kwargs)
Review Comment:
```suggestion
```
Nothing to see here π€¦
##########
airflow/models/trigger.py:
##########
@@ -116,7 +116,15 @@ def _decrypt_kwargs(encrypted_kwargs: str) -> dict[str, Any]:
from airflow.models.crypto import get_fernet
from airflow.serialization.serialized_objects import BaseSerialization
- decrypted_kwargs = json.loads(get_fernet().decrypt(encrypted_kwargs.encode("utf-8")).decode("utf-8"))
+ # We weren't able to encrypt the kwargs in all migration paths,
+ # so we need to handle the case where they are not encrypted.
+ # Triggers aren't long lasting, so we can skip encrypting them now.
+ if encrypted_kwargs.startswith("{"):
+ decrypted_kwargs = json.loads(encrypted_kwargs)
Review Comment:
This fixes the offline upgrade path by detecting the kwargs aren't actually encrypted.
##########
tests/models/test_trigger.py:
##########
@@ -378,3 +380,19 @@ def test_serialize_sensitive_kwargs():
assert isinstance(trigger_row.encrypted_kwargs, str)
assert "value1" not in trigger_row.encrypted_kwargs
assert "value2" not in trigger_row.encrypted_kwargs
+
+
+def test_kwargs_not_encrypted():
+ """
+ Tests that we don't decrypt kwargs if they aren't encrypted.
+ We weren't able to encrypt the kwargs in all migration paths.
+ """
+ trigger = Trigger(classpath="airflow.triggers.testing.SuccessTrigger", kwargs={})
+ # force the `encrypted_kwargs` to be unencrypted, like they would be after an upgrade
Review Comment:
```suggestion
# force the `encrypted_kwargs` to be unencrypted, like they would be after an offline upgrade
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1578971118
##########
airflow/migrations/versions/0140_2_9_0_update_trigger_kwargs_type.py:
##########
@@ -38,13 +41,43 @@
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():
- """Update trigger kwargs type to string"""
+ """Update trigger kwargs type to string and encrypt"""
with op.batch_alter_table("trigger") as batch_op:
batch_op.alter_column("kwargs", type_=sa.Text(), )
+ if not context.is_offline_mode():
+ session = get_session()
+ try:
+ for trigger in session.query(Trigger):
+ trigger.kwargs = trigger.kwargs
+ session.commit()
+ finally:
+ session.close()
+
def downgrade():
- """Unapply update trigger kwargs type to string"""
+ """Unapply update trigger kwargs type to string and encrypt"""
+ if context.is_offline_mode():
+ print(dedent("""
+ ------------
+ -- WARNING: Unable to decrypt trigger kwargs automatically in offline mode!
Review Comment:
This is what it looks like in the generated sql:
```
BEGIN;
-- Running downgrade 1949afb29106 -> ee1467d4aa35
------------
-- WARNING: Unable to decrypt trigger kwargs automatically in offline mode!
-- If any trigger rows exist when you do an offline downgrade, the migration will fail.
------------
ALTER TABLE trigger ALTER COLUMN kwargs TYPE JSON USING kwargs::json;
UPDATE alembic_version SET version_num='ee1467d4aa35' WHERE alembic_version.version_num = '1949afb29106';
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1579935696
##########
airflow/models/trigger.py:
##########
@@ -116,7 +116,15 @@ def _decrypt_kwargs(encrypted_kwargs: str) -> dict[str, Any]:
from airflow.models.crypto import get_fernet
from airflow.serialization.serialized_objects import BaseSerialization
- decrypted_kwargs = json.loads(get_fernet().decrypt(encrypted_kwargs.encode("utf-8")).decode("utf-8"))
+ # We weren't able to encrypt the kwargs in all migration paths,
+ # so we need to handle the case where they are not encrypted.
+ # Triggers aren't long lasting, so we can skip encrypting them now.
+ if encrypted_kwargs.startswith("{"):
+ decrypted_kwargs = json.loads(encrypted_kwargs)
Review Comment:
i kindof like EAFP better.... i guess i'm using that acronym now...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix trigger kwarg encryption migration [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #39246:
URL: https://github.com/apache/airflow/pull/39246#discussion_r1579957208
##########
airflow/migrations/versions/0140_2_9_0_update_trigger_kwargs_type.py:
##########
@@ -38,13 +42,43 @@
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():
- """Update trigger kwargs type to string"""
+ """Update trigger kwargs type to string and encrypt"""
with op.batch_alter_table("trigger") as batch_op:
batch_op.alter_column("kwargs", type_=sa.Text(), )
+ if not context.is_offline_mode():
Review Comment:
Yep. I generally like doing exactly that, not sure why I didn't this time around.
At this point, I'm going to save it for a future refactor PR though.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org