fixup! Only allow webserver to request from the worker log server

airflow/utils/log/file_task_handler.py

@@ -177,9 +177,13 @@ def _read(self, ti, try_number, metadata=None):
                     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',
                 )
 
-                response = httpx.get(url, timeout=timeout, headers={'Authorization': signer.dumps({})})
+                response = httpx.get(
+                    url, timeout=timeout, headers={'Authorization': signer.dumps(log_relative_path)}
+                )
                 response.encoding = "utf-8"
 
                 # Check if the resource was properly fetched

airflow/utils/serve_logs.py

@@ -35,6 +35,8 @@ def flask_app():
         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',
     )
 
     # Prevent direct access to the logs port
@@ -45,13 +47,16 @@ def validate_pre_signed_url():
 
             # We don't actually care about the payload, just that the signature
             # was valid and the `exp` claim is correct
-            _, headers = signer.loads(auth, return_header=True)
+            filename, headers = signer.loads(auth, return_header=True)
 
             issued_at = int(headers['iat'])
             expires_at = int(headers['exp'])
-        except Exception as e:
-            print(e)
+        except Exception:
             abort(403)
+
+        if filename != request.view_args['filename']:
+            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:

tests/utils/test_serve_logs.py

@@ -45,57 +45,66 @@ def sample_log(tmpdir):
     return f
 
 
+@pytest.fixture
+def signer():
+    return TimedJSONWebSignatureSerializer(
+        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',
+    )
+
+
 @pytest.mark.usefixtures('sample_log')
 class TestServeLogs:
     def test_forbidden_no_auth(self, client: "FlaskClient"):
         assert 403 == client.get('/log/sample.log').status_code
 
-    def test_should_serve_file(self, client: "FlaskClient"):
-        signer = TimedJSONWebSignatureSerializer(
-            secret_key=conf.get('webserver', 'secret_key'),
-            algorithm_name='HS512',
-            expires_in=30,
-        )
+    def test_should_serve_file(self, client: "FlaskClient", signer):
         assert (
             LOG_DATA
             == client.get(
                 '/log/sample.log',
                 headers={
-                    'Authorization': signer.dumps({}),
+                    'Authorization': signer.dumps('sample.log'),
                 },
             ).data.decode()
         )
 
-    def test_forbidden_too_long_validity(self, client: "FlaskClient"):
-        signer = TimedJSONWebSignatureSerializer(
-            secret_key=conf.get('webserver', 'secret_key'),
-            algorithm_name='HS512',
-            expires_in=3600,
-        )
+    def test_forbidden_too_long_validity(self, client: "FlaskClient", signer):
+        signer.expires_in = 3600
         assert (
             403
             == client.get(
                 '/log/sample.log',
                 headers={
-                    'Authorization': signer.dumps({}),
+                    'Authorization': signer.dumps('sample.log'),
                 },
             ).status_code
         )
 
-    def test_forbidden_expired(self, client: "FlaskClient"):
-        signer = TimedJSONWebSignatureSerializer(
-            secret_key=conf.get('webserver', 'secret_key'),
-            algorithm_name='HS512',
-            expires_in=30,
-        )
+    def test_forbidden_expired(self, client: "FlaskClient", signer):
         # Fake the time we think we are
         signer.now = lambda: 0
         assert (
             403
             == client.get(
                 '/log/sample.log',
                 headers={
-                    'Authorization': signer.dumps({}),
+                    'Authorization': signer.dumps('sample.log'),
+                },
+            ).status_code
+        )
+
+    def test_wrong_context(self, client: "FlaskClient", signer):
+        signer.salt = None
+        assert (
+            403
+            == client.get(
+                '/log/sample.log',
+                headers={
+                    'Authorization': signer.dumps('sample.log'),
                 },
             ).status_code
         )