Fix auth

airflow/www/app.py

@@ -44,6 +44,7 @@
 from airflow.www.extensions.init_manifest_files import configure_manifest_files
 from airflow.www.extensions.init_robots import init_robots
 from airflow.www.extensions.init_security import (
+    init_api_auth,
     init_cache_control,
     init_check_user_active,
     init_xframe_protection,
@@ -148,6 +149,8 @@ def create_app(config=None, testing=False):
 
     init_dagbag(flask_app)
 
+    init_api_auth(flask_app)
+
     init_robots(flask_app)
 
     init_cache(flask_app)

airflow/www/extensions/init_security.py

@@ -17,10 +17,12 @@
 from __future__ import annotations
 
 import logging
+from importlib import import_module
 
 from flask import redirect, request
 
 from airflow.configuration import conf
+from airflow.exceptions import AirflowConfigException, AirflowException
 from airflow.www.extensions.init_auth_manager import get_auth_manager
 
 log = logging.getLogger(__name__)
@@ -45,6 +47,25 @@ def apply_caching(response):
     app.after_request(apply_caching)
 
 
+def init_api_auth(app):
+    """Load authentication backends."""
+    auth_backends = "airflow.api.auth.backend.default"
+    try:
+        auth_backends = conf.get("api", "auth_backends")
+    except AirflowConfigException:
+        pass
+
+    app.api_auth = []
+    try:
+        for backend in auth_backends.split(","):
+            auth = import_module(backend.strip())
+            auth.init_app(app)
+            app.api_auth.append(auth)
+    except ImportError as err:
+        log.critical("Cannot import %s for API authentication due to: %s", backend, err)
+        raise AirflowException(err)
+
+
 def init_cache_control(app):
     def apply_cache_control(response):
         if "Cache-Control" not in response.headers:

tests/api_connexion/conftest.py

@@ -28,6 +28,7 @@ def minimal_app_for_api():
     @dont_initialize_flask_app_submodules(
         skip_all_except=[
             "init_appbuilder",
+            "init_api_auth",
             "init_api_connexion",
             "init_api_error_handlers",
             "init_airflow_session_interface",

tests/providers/google/common/auth_backend/__init__.py

@@ -0,0 +1,16 @@
+# 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.

tests/providers/google/common/auth_backend/test_google_openid.py

@@ -0,0 +1,147 @@
+# 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 __future__ import annotations
+
+from unittest import mock
+
+import pytest
+from google.auth.exceptions import GoogleAuthError
+
+from airflow.www.app import create_app
+from tests.test_utils.config import conf_vars
+from tests.test_utils.db import clear_db_pools
+from tests.test_utils.decorators import dont_initialize_flask_app_submodules
+
+
+@pytest.fixture(scope="module")
+def google_openid_app():
+    @dont_initialize_flask_app_submodules(
+        skip_all_except=[
+            "init_appbuilder",
+            "init_api_connexion",
+            "init_api_error_handlers",
+            "init_airflow_session_interface",
+            "init_appbuilder_views",
+        ]
+    )
+    def factory():
+        with conf_vars(
+            {("api", "auth_backends"): "airflow.providers.google.common.auth_backend.google_openid"}
+        ):
+            _app = create_app(testing=True, config={"WTF_CSRF_ENABLED": False})  # type:ignore
+            _app.config["AUTH_ROLE_PUBLIC"] = None
+            return _app
+
+    return factory()
+
+
+@pytest.fixture(scope="module")
+def admin_user(google_openid_app):
+    appbuilder = google_openid_app.appbuilder
+    role_admin = appbuilder.sm.find_role("Admin")
+    tester = appbuilder.sm.find_user(username="test")
+    if not tester:
+        appbuilder.sm.add_user(
+            username="test",
+            first_name="test",
+            last_name="test",
+            email="[email protected]",
+            role=role_admin,
+            password="test",
+        )
+    return role_admin
+
+
+@pytest.mark.skip_if_database_isolation_mode
+@pytest.mark.db_test
+class TestGoogleOpenID:
+    @pytest.fixture(autouse=True)
+    def _set_attrs(self, google_openid_app, admin_user) -> None:
+        self.app = google_openid_app
+        self.admin_user = admin_user
+
+    @mock.patch("google.oauth2.id_token.verify_token")
+    def test_success(self, mock_verify_token):
+        clear_db_pools()
+        mock_verify_token.return_value = {
+            "iss": "accounts.google.com",
+            "email_verified": True,
+            "email": "[email protected]",
+        }
+
+        with self.app.test_client() as test_client:
+            response = test_client.get("/api/v1/pools", headers={"Authorization": "bearer JWT_TOKEN"})
+
+            assert 200 == response.status_code
+            assert "Default pool" in str(response.json)
+
+    @pytest.mark.parametrize("auth_header", ["bearer", "JWT_TOKEN", "bearer "])
+    @mock.patch("google.oauth2.id_token.verify_token")
+    def test_malformed_headers(self, mock_verify_token, auth_header):
+        mock_verify_token.return_value = {
+            "iss": "accounts.google.com",
+            "email_verified": True,
+            "email": "[email protected]",
+        }
+
+        with self.app.test_client() as test_client:
+            response = test_client.get("/api/v1/pools", headers={"Authorization": auth_header})
+
+        assert 401 == response.status_code
+
+    @mock.patch("google.oauth2.id_token.verify_token")
+    def test_invalid_iss_in_jwt_token(self, mock_verify_token):
+        mock_verify_token.return_value = {
+            "iss": "INVALID",
+            "email_verified": True,
+            "email": "[email protected]",
+        }
+
+        with self.app.test_client() as test_client:
+            response = test_client.get("/api/v1/pools", headers={"Authorization": "bearer JWT_TOKEN"})
+
+        assert 401 == response.status_code
+
+    @mock.patch("google.oauth2.id_token.verify_token")
+    def test_user_not_exists(self, mock_verify_token):
+        mock_verify_token.return_value = {
+            "iss": "accounts.google.com",
+            "email_verified": True,
+            "email": "[email protected]",
+        }
+
+        with self.app.test_client() as test_client:
+            response = test_client.get("/api/v1/pools", headers={"Authorization": "bearer JWT_TOKEN"})
+
+        assert 401 == response.status_code
+
+    @conf_vars({("api", "auth_backends"): "airflow.providers.google.common.auth_backend.google_openid"})
+    def test_missing_id_token(self):
+        with self.app.test_client() as test_client:
+            response = test_client.get("/api/v1/pools")
+
+        assert 401 == response.status_code
+
+    @conf_vars({("api", "auth_backends"): "airflow.providers.google.common.auth_backend.google_openid"})
+    @mock.patch("google.oauth2.id_token.verify_token")
+    def test_invalid_id_token(self, mock_verify_token):
+        mock_verify_token.side_effect = GoogleAuthError("Invalid token")
+
+        with self.app.test_client() as test_client:
+            response = test_client.get("/api/v1/pools", headers={"Authorization": "bearer JWT_TOKEN"})
+
+        assert 401 == response.status_code