破坏的DAG：[/ airflow / dags / a.py]无法为login = None解密`extra`参数，缺少FERNET_KEY配置

Question

我开始没有FERNET_KEY的Airflow。一旦我意识到，我做了以下：https://airflow.apache.org/configuration.html#connections

pip install apache-airflow[crypto]

from cryptography.fernet import Fernet
fernet_key= Fernet.generate_key()
print(fernet_key)

拿了钥匙把它放在airflow.cfg然后叫airflow initdb，但错误仍然出现。

我究竟做错了什么？

当我做：

airflow webserver -D

我明白了：

  File "/usr/local/lib/python2.7/dist-packages/airflow/models.py", line 713, in extra_dejson
    if self.extra:
  File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/attributes.py", line 293, in __get__
    return self.descriptor.__get__(instance, owner)
  File "/usr/local/lib/python2.7/dist-packages/airflow/models.py", line 632, in get_extra
    return fernet.decrypt(bytes(self._extra, 'utf-8')).decode()
  File "/usr/lib/python2.7/dist-packages/cryptography/fernet.py", line 101, in decrypt
    raise InvalidToken

日志表明此代码存在问题：

def get_conn(conn_id, session=None):
    conn = (session.query(Connection)
                   .filter(Connection.conn_id == conn_id)
                   .first())
    return conn


def my_python_function():
   conn = get_conn('s3connection')
   key_id = conn.extra_dejson.get('aws_access_key_id')
   secret_key = conn.extra_dejson.get('aws_secret_access_key')
   default_region = conn.extra_dejson.get('region_name')
   return key_id,secret_key,default_region

Answer 1

气流通常会为您生成一个。

这是一个例子：

$ python
>>> from cryptography.fernet import Fernet
>>> k=Fernet.generate_key()
>>> print(k)
Z6BkzaWcF7r5cC-VMAumjpBpudSyjGskQ0ObquGJhG0=
>>> ^D
$ $EDITOR $AIRFLOW_HOME/airflow.cfg

有变化：

# Secret key to save connection passwords in the db
fernet_key = cryptography_not_found_storing_passwords_in_plain_text

至：

# Secret key to save connection passwords in the db
fernet_key = Z6BkzaWcF7r5cC-VMAumjpBpudSyjGskQ0ObquGJhG0=

检查它是否按预期设置（或者每次都会生成一个随机的）

$ python
Python 2.7.13 (default, Jul 18 2017, 09:17:00)
[GCC 4.2.1 Compatible Apple LLVM 8.1.0 (clang-802.0.42)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> from airflow import configuration as conf
[2018-06-14 17:53:36,200] {__init__.py:57} INFO - Using executor SequentialExecutor
>>> conf.get('core','fernet_key')
'Z6BkzaWcF7r5cC-VMAumjpBpudSyjGskQ0ObquGJhG0='
>>>

上面应该是v1.9.0和v1.8.2语法[固定]，我用后者对此进行了双重检查。

每当您更改您的fernet密钥时，您需要删除所有使用加密的连接和变量，因为它们不再解密。

你可以重置你的数据库，但这可能是过度的。

Answer 2

使用Fernet，Airflow会加密后端数据库中连接的所有密码。

在您的情况下，Airflow后端正在使用以前的fernet密钥，您已经生成了一个用于创建新连接的密钥。

我的建议是做以下几点，

airflow resetdb

这将有助于删除后端数据库中的所有现有记录。

然后，

airflow initdb

这会像新鲜一样初始化后端。

然后启动气流Web服务器和调度程序

airflow web server -p {port}
airflow scheduler

然后在UI中创建s3的新连接（在额外 - {“aws_access_key_id”：“_ your_aws_access_key_id_”，“aws_secret_access_key”：“_ your_aws_secret_access_key_”}）

现在，您应该能够通过以下解决方案测试s3文件观察器 - Airflow s3 connection using UI

Answer 3

如果您在没有fernet密钥的情况下启动气流，则不允许对任何连接进行加密。

如果您有一个fernet密钥，添加/编辑了任何连接，它们将使用该密钥加密。如果以任何方式更改密钥，则这些连接将无法解密。如果您当前使用不同的fernet密钥加密连接，唯一的解决方案是将使用一个密钥创建的密钥迁移到另一个密钥。或者只是删除它们并重新创建它们，假设您仍然可以使用值。