As of right now I am trying to utilize Airflow's SimpleHTTPOperator and would like to use endpoints that require jwt authentication. Right now I am using Node.js and Passport.js to authenticate endpoints. I store the jwt within a cookie using res.cookie after hitting a login page and it stores the jwt within the cookie correctly after logging in. This is the cookie header parameters within Postman.
Key: Cookie
Value:
connect.sid=[CONNECTION ID NOW STORED HERE];
jwt=[JWT NOW STORED HERE]
And whenever I attempt to access endpoints that require authentication it works properly within Postman.
So when building out my DAG definition file, I create a task to login first so that the other operators can access the JWT and allow the endpoints to run. t1, t2, and t3 require jwt authentication by reading the headers and checking the Cookie. Here is my Python code using the SimpleHTTPOperators:
from datetime import datetime, timedelta
from airflow.operators.http_operator import SimpleHttpOperator
from airflow import DAG
default_args = {
'owner': 'what',
'depends_on_past': False,
'start_date': datetime(2020, 8, 12),
'email': 'what@gmail.com',
'email_on_failure': False,
'email_on_retry': False,
'retries': 3,
'retry_delay': timedelta(days=1),
'concurrency': 2,
}
# PT is 7 hours behind UTC, this will be Noon PT
dag = DAG('services',
schedule_interval='0 19 * * *',
default_args=default_args)
t0 = SimpleHttpOperator(
task_id='login',
method='POST',
http_conn_id='http_service',
endpoint='/account_service/login?email=admin@example.com&password=admin',
dag=dag)
t1 = SimpleHttpOperator(
task_id='delete_comments',
method='POST',
http_conn_id='http_service',
endpoint='/comment_service/delete_comments',
dag=dag)
t2 = SimpleHttpOperator(
task_id='delete_photos',
method='POST',
http_conn_id='http_service',
endpoint='/photo_service/delete_photos',
dag=dag)
t3 = SimpleHttpOperator(
task_id='pending_to_deleted_account',
method='POST',
http_conn_id='http_service',
endpoint='/account_service/pending_to_deleted',
dag=dag)
t0 >> t2 >> t3
t0 >> t1 >> t3
However, when I try to call various endpoints that need authentication using Airflow in Python, it ends up scheduling for retrying due to 401 errors.
[2020-08-25 14:22:12,050] {logging_mixin.py:112} INFO - [2020-08-25 14:22:12,049] {base_hook.py:87} INFO - Using connection to: id: http_service. Host: localhost, Port: 8000, Schema: None, Login: None, Password: None, extra: None
[2020-08-25 14:22:12,052] {logging_mixin.py:112} INFO - [2020-08-25 14:22:12,051] {http_hook.py:136} INFO - Sending 'POST' to url: http://localhost:8000/comment_service/delete_comments
[2020-08-25 14:22:12,056] {logging_mixin.py:112} INFO - [2020-08-25 14:22:12,056] {http_hook.py:150} ERROR - HTTP error: Unauthorized
[2020-08-25 14:22:12,057] {logging_mixin.py:112} INFO - [2020-08-25 14:22:12,057] {http_hook.py:151} ERROR - Unauthorized
[2020-08-25 14:22:12,067] {taskinstance.py:1145} ERROR - 401:Unauthorized
Traceback (most recent call last):
File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/airflow/hooks/http_hook.py", line 148, in check_response
response.raise_for_status()
File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/requests/models.py", line 941, in raise_for_status
raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 401 Client Error: Unauthorized for url: http://localhost:8000/comment_service/delete_comments
Is there any way to let the other operators access response header values such as the cookie stored from the login so that the authenticated endpoints can recognize the cookie and run properly?
EDIT 1: I figured out a very hacky way to bypass this issue by going into the Connections tab and placing in the following parameters:
Conn Id: http_service
Port: 8000
Extra: { "Cookie": "jwt=[MY JWT HERE]"}
I get a jwt by calling the POST login route and look into the headers within Postman and place it into where the [MY JWT HERE] slot is placed. While this works, this does not solve my problem as I would like to instead read the header response given from the t0 SimpleHTTPOperator. I have tried looking into getting responses by looking at the XCom documentation of Airflow, but the only progress I made was looking into the execute function like so:
print(t0.execute(context))
which simply results in printing out only the body of the response rather than multiple parameters such as the headers, body, etc.
Here is the source code of the HTTP operator execute function:
Any help would be appreciated!