Improve security and error handling for the internal API

airflow/api_internal/endpoints/rpc_api_endpoint.py

@@ -163,6 +163,12 @@ def log_and_build_error_response(message, status):
 
 def internal_airflow_api(body: dict[str, Any]) -> APIResponse:
     """Handle Internal API /internal_api/v1/rpcapi endpoint."""
+    content_type = request.headers.get("Content-Type")
+    if content_type != "application/json":
+        raise PermissionDenied("Expected Content-Type: application/json")
+    accept = request.headers.get("Accept")
+    if accept != "application/json":
+        raise PermissionDenied("Expected Accept: application/json")
     auth = request.headers.get("Authorization", "")
     signer = JWTSigner(
         secret_key=conf.get("core", "internal_api_secret_key"),
@@ -177,11 +183,11 @@ def internal_airflow_api(body: dict[str, Any]) -> APIResponse:
     except BadSignature:
         raise PermissionDenied("Bad Signature. Please use only the tokens provided by the API.")
     except InvalidAudienceError:
-        raise PermissionDenied("Invalid audience for the request", exc_info=True)
+        raise PermissionDenied("Invalid audience for the request")
     except InvalidSignatureError:
-        raise PermissionDenied("The signature of the request was wrong", exc_info=True)
+        raise PermissionDenied("The signature of the request was wrong")
     except ImmatureSignatureError:
-        raise PermissionDenied("The signature of the request was sent from the future", exc_info=True)
+        raise PermissionDenied("The signature of the request was sent from the future")
     except ExpiredSignatureError:
         raise PermissionDenied(
             "The signature of the request has expired. Make sure that all components "

airflow/api_internal/internal_api_call.py

@@ -119,6 +119,7 @@ def make_jsonrpc_request(method_name: str, params_json: str) -> bytes:
         )
         headers = {
             "Content-Type": "application/json",
+            "Accept": "application/json",
             "Authorization": signer.generate_signed_token({"method": method_name}),
         }
         data = {"jsonrpc": "2.0", "method": method_name, "params": params_json}

airflow/www/extensions/init_auth_manager.py

@@ -65,3 +65,8 @@ def get_auth_manager() -> BaseAuthManager:
             "The `init_auth_manager` method needs to be called first."
         )
     return auth_manager
+
+
+def is_auth_manager_initialized() -> bool:
+    """Return whether the auth manager has been initialized."""
+    return auth_manager is not None

airflow/www/views.py

@@ -137,7 +137,7 @@
 from airflow.version import version
 from airflow.www import auth, utils as wwwutils
 from airflow.www.decorators import action_logging, gzipped
-from airflow.www.extensions.init_auth_manager import get_auth_manager
+from airflow.www.extensions.init_auth_manager import get_auth_manager, is_auth_manager_initialized
 from airflow.www.forms import (
     DagRunEditForm,
     DateTimeForm,
@@ -688,6 +688,9 @@ def method_not_allowed(error):
 
 def show_traceback(error):
     """Show Traceback for a given error."""
+    if not is_auth_manager_initialized():
+        # this is the case where internal API component is used and auth manager is not used/initialized
+        return ("Error calling the API", 500)
     is_logged_in = get_auth_manager().is_logged_in()
     return (
         render_template(

tests/api_internal/endpoints/test_rpc_api_endpoint.py

@@ -123,6 +123,7 @@ def test_method(self, input_params, method_result, result_cmp_func, method_param
         mock_test_method.return_value = method_result
         headers = {
             "Content-Type": "application/json",
+            "Accept": "application/json",
             "Authorization": signer.generate_signed_token({"method": TEST_METHOD_NAME}),
         }
         input_data = {
@@ -148,6 +149,7 @@ def test_method(self, input_params, method_result, result_cmp_func, method_param
     def test_method_with_exception(self, signer: JWTSigner):
         headers = {
             "Content-Type": "application/json",
+            "Accept": "application/json",
             "Authorization": signer.generate_signed_token({"method": TEST_METHOD_NAME}),
         }
         mock_test_method.side_effect = ValueError("Error!!!")
@@ -162,6 +164,7 @@ def test_unknown_method(self, signer: JWTSigner):
         UNKNOWN_METHOD = "i-bet-it-does-not-exist"
         headers = {
             "Content-Type": "application/json",
+            "Accept": "application/json",
             "Authorization": signer.generate_signed_token({"method": UNKNOWN_METHOD}),
         }
         data = {"jsonrpc": "2.0", "method": UNKNOWN_METHOD, "params": {}}
@@ -174,6 +177,7 @@ def test_unknown_method(self, signer: JWTSigner):
     def test_invalid_jsonrpc(self, signer: JWTSigner):
         headers = {
             "Content-Type": "application/json",
+            "Accept": "application/json",
             "Authorization": signer.generate_signed_token({"method": TEST_METHOD_NAME}),
         }
         data = {"jsonrpc": "1.0", "method": TEST_METHOD_NAME, "params": {}}
@@ -194,13 +198,14 @@ def test_missing_token(self):
         with pytest.raises(PermissionDenied, match="Unable to authenticate API via token."):
             self.client.post(
                 "/internal_api/v1/rpcapi",
-                headers={"Content-Type": "application/json"},
+                headers={"Content-Type": "application/json", "Accept": "application/json"},
                 data=json.dumps(input_data),
             )
 
     def test_invalid_token(self, signer: JWTSigner):
         headers = {
             "Content-Type": "application/json",
+            "Accept": "application/json",
             "Authorization": signer.generate_signed_token({"method": "WRONG_METHOD_NAME"}),
         }
         data = {"jsonrpc": "1.0", "method": TEST_METHOD_NAME, "params": {}}
@@ -209,3 +214,24 @@ def test_invalid_token(self, signer: JWTSigner):
             PermissionDenied, match="Bad Signature. Please use only the tokens provided by the API."
         ):
             self.client.post("/internal_api/v1/rpcapi", headers=headers, data=json.dumps(data))
+
+    def test_missing_accept(self, signer: JWTSigner):
+        headers = {
+            "Content-Type": "application/json",
+            "Authorization": signer.generate_signed_token({"method": "WRONG_METHOD_NAME"}),
+        }
+        data = {"jsonrpc": "1.0", "method": TEST_METHOD_NAME, "params": {}}
+
+        with pytest.raises(PermissionDenied, match="Expected Accept: application/json"):
+            self.client.post("/internal_api/v1/rpcapi", headers=headers, data=json.dumps(data))
+
+    def test_wrong_accept(self, signer: JWTSigner):
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/html",
+            "Authorization": signer.generate_signed_token({"method": "WRONG_METHOD_NAME"}),
+        }
+        data = {"jsonrpc": "1.0", "method": TEST_METHOD_NAME, "params": {}}
+
+        with pytest.raises(PermissionDenied, match="Expected Accept: application/json"):
+            self.client.post("/internal_api/v1/rpcapi", headers=headers, data=json.dumps(data))