This is an automated email from the ASF dual-hosted git repository. potiuk pushed a commit to branch v2-3-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 50657119da418366b91d2da180551b852be4b399 Author: Jarek Potiuk <[email protected]> AuthorDate: Sat Jun 18 23:08:51 2022 +0200 Get rid of TimedJSONWebSignatureSerializer (#24519) The TimedJSONWebSignatureSerializer has been deprecated from the itsdangerous library and they recommended to use dedicated libraries for it. https://github.com/pallets/itsdangerous/issues/129 Since we are going to move to FAB 4+ with #22397 where newer version of itsdangerous is used, we need to switch to another library. We are already using PyJWT so the choice is obvious. Additionally to switching, the following improvements were done: * the use of JWT claims has been fixed to follow JWT standard. We were using "iat" header wrongly. The specification of JWT only expects the header to be there and be valid UTC timestamp, but the claim does not impact maturity of the signature - the signature is valid if iat is in the future. Instead "nbf" - "not before" claim should be used to verify if the request is not coming from the future. We now require all claims to be present in the request. * rather than using salt/signing_context we switched to standard JWT "audience" claim (same end result) * we have now much better diagnostics on the server side of the reason why request is forbidden - explicit error messages are printed in server logs and details of the exception. This is secure, we do not spill the information about the reason to the client, it's only available in server logs, so there is no risk attacker could use it. * the JWTSigner is "use-agnostic". We should be able to use the same class for any other signatures (Internal API from AIP-44) with just different audience * Short, 5 seconds default clock skew is allowed, to account for systems that have "almost" synchronized time * more tests addded with proper time freezing testing both expiry and immaturity of the request This change is not a breaking one because the JWT authentication details are not "public API" - but in case someone reverse engineered our claims and implemented their own log file retrieval, we should add a change in our changelog - therefore newsfragment is added. (cherry picked from commit 1f8e4c955394b31462956501d9a6741b98892f3a) --- airflow/utils/jwt_signer.py | 82 ++++++++++++++++++ airflow/utils/log/file_task_handler.py | 18 ++-- airflow/utils/serve_logs.py | 84 +++++++++++++------ newsfragments/24519.misc.rst | 1 + tests/utils/test_serve_logs.py | 147 +++++++++++++++++++++++++++------ 5 files changed, 272 insertions(+), 60 deletions(-) diff --git a/airflow/utils/jwt_signer.py b/airflow/utils/jwt_signer.py new file mode 100644 index 0000000000..941a3d0598 --- /dev/null +++ b/airflow/utils/jwt_signer.py @@ -0,0 +1,82 @@ +# 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. +from datetime import datetime, timedelta +from typing import Any, Dict + +import jwt + + +class JWTSigner: + """ + Signs and verifies JWT Token. Used to authorise and verify requests. + + :param secret_key: key used to sign the request + :param expiration_time_in_seconds: time after which the token becomes invalid (in seconds) + :param audience: audience that the request is expected to have + :param leeway_in_seconds: leeway that allows for a small clock skew between the two parties + :param algorithm: algorithm used for signing + """ + + def __init__( + self, + secret_key: str, + expiration_time_in_seconds: int, + audience: str, + leeway_in_seconds: int = 5, + algorithm: str = "HS512", + ): + self._secret_key = secret_key + self._expiration_time_in_seconds = expiration_time_in_seconds + self._audience = audience + self._leeway_in_seconds = leeway_in_seconds + self._algorithm = algorithm + + def generate_signed_token(self, extra_payload: Dict[str, Any]) -> str: + """ + Generate JWT with extra payload added. + :param extra_payload: extra payload that is added to the signed token + :return: signed token + """ + jwt_dict = { + "aud": self._audience, + "iat": datetime.utcnow(), + "nbf": datetime.utcnow(), + "exp": datetime.utcnow() + timedelta(seconds=self._expiration_time_in_seconds), + } + jwt_dict.update(extra_payload) + token = jwt.encode( + jwt_dict, + self._secret_key, + algorithm=self._algorithm, + ) + return token + + def verify_token(self, token: str) -> Dict[str, Any]: + payload = jwt.decode( + token, + self._secret_key, + leeway=timedelta(seconds=self._leeway_in_seconds), + algorithms=[self._algorithm], + options={ + "verify_signature": True, + "require_exp": True, + "require_iat": True, + "require_nbf": True, + }, + audience=self._audience, + ) + return payload diff --git a/airflow/utils/log/file_task_handler.py b/airflow/utils/log/file_task_handler.py index e0561991b2..321e125288 100644 --- a/airflow/utils/log/file_task_handler.py +++ b/airflow/utils/log/file_task_handler.py @@ -22,11 +22,10 @@ from datetime import datetime from pathlib import Path from typing import TYPE_CHECKING, Optional, Tuple -from itsdangerous import TimedJSONWebSignatureSerializer - from airflow.configuration import AirflowConfigException, conf from airflow.utils.context import Context from airflow.utils.helpers import parse_template_string, render_template_to_string +from airflow.utils.jwt_signer import JWTSigner from airflow.utils.log.non_caching_file_handler import NonCachingFileHandler if TYPE_CHECKING: @@ -191,16 +190,17 @@ class FileTaskHandler(logging.Handler): except (AirflowConfigException, ValueError): pass - signer = TimedJSONWebSignatureSerializer( + signer = JWTSigner( secret_key=conf.get('webserver', 'secret_key'), - algorithm_name='HS512', - expires_in=conf.getint('webserver', 'log_request_clock_grace', fallback=30), - # This isn't really a "salt", more of a signing context - salt='task-instance-logs', + expiration_time_in_seconds=conf.getint( + 'webserver', 'log_request_clock_grace', fallback=30 + ), + audience="task-instance-logs", ) - response = httpx.get( - url, timeout=timeout, headers={'Authorization': signer.dumps(log_relative_path)} + url, + timeout=timeout, + headers={b'Authorization': signer.generate_signed_token({"filename": log_relative_path})}, ) response.encoding = "utf-8" diff --git a/airflow/utils/serve_logs.py b/airflow/utils/serve_logs.py index 50fdb47a02..e14162178b 100644 --- a/airflow/utils/serve_logs.py +++ b/airflow/utils/serve_logs.py @@ -16,55 +16,89 @@ # under the License. """Serve logs process""" +import logging import os -import time import gunicorn.app.base from flask import Flask, abort, request, send_from_directory -from itsdangerous import TimedJSONWebSignatureSerializer +from jwt.exceptions import ( + ExpiredSignatureError, + ImmatureSignatureError, + InvalidAudienceError, + InvalidIssuedAtError, + InvalidSignatureError, +) from setproctitle import setproctitle from airflow.configuration import conf +from airflow.utils.docs import get_docs_url +from airflow.utils.jwt_signer import JWTSigner + +logger = logging.getLogger(__name__) def create_app(): flask_app = Flask(__name__, static_folder=None) - max_request_age = conf.getint('webserver', 'log_request_clock_grace', fallback=30) + expiration_time_in_seconds = conf.getint('webserver', 'log_request_clock_grace', fallback=30) log_directory = os.path.expanduser(conf.get('logging', 'BASE_LOG_FOLDER')) - signer = TimedJSONWebSignatureSerializer( + signer = JWTSigner( secret_key=conf.get('webserver', 'secret_key'), - algorithm_name='HS512', - expires_in=max_request_age, - # This isn't really a "salt", more of a signing context - salt='task-instance-logs', + expiration_time_in_seconds=expiration_time_in_seconds, + audience="task-instance-logs", ) # Prevent direct access to the logs port @flask_app.before_request def validate_pre_signed_url(): try: - auth = request.headers['Authorization'] - - # We don't actually care about the payload, just that the signature - # was valid and the `exp` claim is correct - filename, headers = signer.loads(auth, return_header=True) - - issued_at = int(headers['iat']) - expires_at = int(headers['exp']) - except Exception: + auth = request.headers.get('Authorization') + if auth is None: + logger.warning("The Authorization header is missing: %s.", request.headers) + abort(403) + payload = signer.verify_token(auth) + token_filename = payload.get("filename") + request_filename = request.view_args['filename'] + if token_filename is None: + logger.warning("The payload does not contain 'filename' key: %s.", payload) + abort(403) + if token_filename != request_filename: + logger.warning( + "The payload log_relative_path key is different than the one in token:" + "Request path: %s. Token path: %s.", + request_filename, + token_filename, + ) + abort(403) + except InvalidAudienceError: + logger.warning("Invalid audience for the request", exc_info=True) abort(403) - - if filename != request.view_args['filename']: + except InvalidSignatureError: + logger.warning("The signature of the request was wrong", exc_info=True) abort(403) - - # Validate the `iat` and `exp` are within `max_request_age` of now. - now = int(time.time()) - if abs(now - issued_at) > max_request_age: + except ImmatureSignatureError: + logger.warning("The signature of the request was sent from the future", exc_info=True) abort(403) - if abs(now - expires_at) > max_request_age: + except ExpiredSignatureError: + logger.warning( + "The signature of the request has expired. Make sure that all components " + "in your system have synchronized clocks. " + "See more at %s", + get_docs_url("configurations-ref.html#secret-key"), + exc_info=True, + ) abort(403) - if issued_at > expires_at or expires_at - issued_at > max_request_age: + except InvalidIssuedAtError: + logger.warning( + "The request was issues in the future. Make sure that all components " + "in your system have synchronized clocks. " + "See more at %s", + get_docs_url("configurations-ref.html#secret-key"), + exc_info=True, + ) + abort(403) + except Exception: + logger.warning("Unknown error", exc_info=True) abort(403) @flask_app.route('/log/<path:filename>') diff --git a/newsfragments/24519.misc.rst b/newsfragments/24519.misc.rst new file mode 100644 index 0000000000..799d9141d2 --- /dev/null +++ b/newsfragments/24519.misc.rst @@ -0,0 +1 @@ +The JWT claims in the request to retrieve logs have been standardized: we use "nbf" and "aud" claims for maturity and audience of the requests. Also "filename" payload field is used to keep log name. diff --git a/tests/utils/test_serve_logs.py b/tests/utils/test_serve_logs.py index 168a43a012..f8d3881759 100644 --- a/tests/utils/test_serve_logs.py +++ b/tests/utils/test_serve_logs.py @@ -14,12 +14,15 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +import datetime from typing import TYPE_CHECKING +import jwt import pytest -from itsdangerous import TimedJSONWebSignatureSerializer +from freezegun import freeze_time from airflow.configuration import conf +from airflow.utils.jwt_signer import JWTSigner from airflow.utils.serve_logs import create_app from tests.test_utils.config import conf_vars @@ -47,12 +50,19 @@ def sample_log(tmpdir): @pytest.fixture def signer(): - return TimedJSONWebSignatureSerializer( + return JWTSigner( secret_key=conf.get('webserver', 'secret_key'), - algorithm_name='HS512', - expires_in=30, - # This isn't really a "salt", more of a signing context - salt='task-instance-logs', + expiration_time_in_seconds=30, + audience="task-instance-logs", + ) + + [email protected] +def different_audience(): + return JWTSigner( + secret_key=conf.get('webserver', 'secret_key'), + expiration_time_in_seconds=30, + audience="different-audience", ) @@ -62,49 +72,134 @@ class TestServeLogs: assert 403 == client.get('/log/sample.log').status_code def test_should_serve_file(self, client: "FlaskClient", signer): + response = client.get( + '/log/sample.log', + headers={ + 'Authorization': signer.generate_signed_token({"filename": 'sample.log'}), + }, + ) + assert response.data.decode() == LOG_DATA + assert response.status_code == 200 + + def test_forbidden_different_logname(self, client: "FlaskClient", signer): + response = client.get( + '/log/sample.log', + headers={ + 'Authorization': signer.generate_signed_token({"filename": 'different.log'}), + }, + ) + assert response.status_code == 403 + + def test_forbidden_expired(self, client: "FlaskClient", signer): + with freeze_time("2010-01-14"): + token = signer.generate_signed_token({"filename": 'sample.log'}) + assert ( + client.get( + '/log/sample.log', + headers={ + 'Authorization': token, + }, + ).status_code + == 403 + ) + + def test_forbidden_future(self, client: "FlaskClient", signer): + with freeze_time(datetime.datetime.utcnow() + datetime.timedelta(seconds=3600)): + token = signer.generate_signed_token({"filename": 'sample.log'}) assert ( - LOG_DATA - == client.get( + client.get( '/log/sample.log', headers={ - 'Authorization': signer.dumps('sample.log'), + 'Authorization': token, }, - ).data.decode() + ).status_code + == 403 ) - def test_forbidden_too_long_validity(self, client: "FlaskClient", signer): - signer.expires_in = 3600 + def test_ok_with_short_future_skew(self, client: "FlaskClient", signer): + with freeze_time(datetime.datetime.utcnow() + datetime.timedelta(seconds=1)): + token = signer.generate_signed_token({"filename": 'sample.log'}) assert ( - 403 - == client.get( + client.get( '/log/sample.log', headers={ - 'Authorization': signer.dumps('sample.log'), + 'Authorization': token, }, ).status_code + == 200 ) - def test_forbidden_expired(self, client: "FlaskClient", signer): - # Fake the time we think we are - signer.now = lambda: 0 + def test_ok_with_short_past_skew(self, client: "FlaskClient", signer): + with freeze_time(datetime.datetime.utcnow() - datetime.timedelta(seconds=31)): + token = signer.generate_signed_token({"filename": 'sample.log'}) + assert ( + client.get( + '/log/sample.log', + headers={ + 'Authorization': token, + }, + ).status_code + == 200 + ) + + def test_forbidden_with_long_future_skew(self, client: "FlaskClient", signer): + with freeze_time(datetime.datetime.utcnow() + datetime.timedelta(seconds=10)): + token = signer.generate_signed_token({"filename": 'sample.log'}) + assert ( + client.get( + '/log/sample.log', + headers={ + 'Authorization': token, + }, + ).status_code + == 403 + ) + + def test_forbidden_with_long_past_skew(self, client: "FlaskClient", signer): + with freeze_time(datetime.datetime.utcnow() - datetime.timedelta(seconds=40)): + token = signer.generate_signed_token({"filename": 'sample.log'}) + assert ( + client.get( + '/log/sample.log', + headers={ + 'Authorization': token, + }, + ).status_code + == 403 + ) + + def test_wrong_audience(self, client: "FlaskClient", different_audience): assert ( - 403 - == client.get( + client.get( '/log/sample.log', headers={ - 'Authorization': signer.dumps('sample.log'), + 'Authorization': different_audience.generate_signed_token({"filename": 'sample.log'}), }, ).status_code + == 403 ) - def test_wrong_context(self, client: "FlaskClient", signer): - signer.salt = None + @pytest.mark.parametrize("claim_to_remove", ["iat", "exp", "nbf", "aud"]) + def test_missing_claims(self, claim_to_remove: str, client: "FlaskClient"): + jwt_dict = { + "aud": "task-instance-logs", + "iat": datetime.datetime.utcnow(), + "nbf": datetime.datetime.utcnow(), + "exp": datetime.datetime.utcnow() + datetime.timedelta(seconds=30), + } + del jwt_dict[claim_to_remove] + jwt_dict.update({"filename": 'sample.log'}) + token = jwt.encode( + jwt_dict, + conf.get('webserver', 'secret_key'), + algorithm="HS512", + ) assert ( - 403 - == client.get( + client.get( '/log/sample.log', headers={ - 'Authorization': signer.dumps('sample.log'), + 'Authorization': token, }, ).status_code + == 403 )
