VaultHook AppRole authentication fails when using a conn_uri

[nathadfield]

Apache Airflow version

2.1.3 (latest released)

Operating System

MacOS Big Sur

Versions of Apache Airflow Providers

apache-airflow-providers-hashicorp==2.0.0

Deployment

Astronomer

Deployment details

No response

What happened

When trying to to use the VaultHook with AppRole authentication via a connection defined as a conn_uri, I was unable to establish a connection in a custom operator due the 'role_id' not being provided despite it being an optional argument.

https://github.com/apache/airflow/blob/main/airflow/providers/hashicorp/hooks/vault.py#L124

What you expected to happen

So, with a connection defined as a URI as follows:

http://[ROLE_ID]:[SECRET_ID]@https://[VAULT_URL]?auth_type=approle

I receive the following error when trying to run a task in a custom operator.

[2021-09-07 08:30:36,877] {base.py:79} INFO - Using connection to: id: vault. Host: https://[VAULT_URL], Port: None, Schema: None, Login: [ROLE_ID], Password: ***, extra: {'auth_type': 'approle'}
[2021-09-07 08:30:36,879] {taskinstance.py:1462} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1164, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1282, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1312, in _execute_task
    result = task_copy.execute(context=context)
  File "/usr/local/airflow/plugins/operators/my_vault_operator.py", line 28, in execute
    vaulthook = VaultHook(vault_conn_id=self.vault_conn_id)
  File "/usr/local/lib/python3.7/site-packages/airflow/providers/hashicorp/hooks/vault.py", line 216, in __init__
    radius_port=radius_port,
  File "/usr/local/lib/python3.7/site-packages/airflow/providers/hashicorp/_internal_client/vault_client.py", line 153, in __init__
    raise VaultError("The 'approle' authentication type requires 'role_id'")
hvac.exceptions.VaultError: The 'approle' authentication type requires 'role_id', on None None

Given that the ROLE_ID and SECRET_ID are part of the connection, I was expecting that the hook should retrieve this from self.connection.login.

How to reproduce

1. Create a connection to Vault defined as a conn_uri, e.g. http://[ROLE_ID]:[SECRET_ID]@https://[VAULT_URL]?auth_type=approle
2. Create a simple custom operator as follows:

from airflow.models import BaseOperator
from airflow.providers.hashicorp.hooks.vault import VaultHook


class MyVaultOperator(BaseOperator):

    def __init__(self, vault_conn_id=None, *args, **kwargs):
        super(MyVaultOperator, self).__init__(*args, **kwargs)
        self.vault_conn_id = vault_conn_id

    def execute(self, context):
        vaulthook = VaultHook(vault_conn_id=self.vault_conn_id)

3. Create a simple DAG to use this operator

from datetime import datetime
from airflow import models
from operators.my_vault_operator import MyVaultOperator

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2011, 1, 1),
}

dag_name = 'my_vault_dag'

with models.DAG(
    dag_name,
    default_args=default_args
) as dag:

    my_vault_task = MyVaultOperator(
        task_id='vault_task',
        vault_conn_id='vault',
    )

4. Run this DAG via airflow tasks test my_vault_dag vault_task 2021-09-06

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[mik-laj]

• Yes I am willing to submit a PR!

I am looking forward to your contribution. I assigned this ticket to you.

[nathadfield]

I think it should be as simple as changing this to this:

if auth_type in ["approle", "aws_iam"]:
    if not role_id:
        if self.connection.login:
            role_id = self.connection.login
        else:
            role_id = self.connection.extra_dejson.get('role_id')

[nathadfield]

Actually this breaks some existing tests because the mocked connection contains the value user for connection.login.

https://github.com/apache/airflow/blob/main/tests/providers/hashicorp/hooks/test_vault.py#L32