Make the role assigned to anonymous users customizable (#14042)

Fixes the issue wherein regardless of what role anonymous users are assigned (via the `AUTH_ROLE_PUBLIC` env var), they can't see any DAGs.

Current behavior causes:
Anonymous users are handled as a special case by Airflow's DAG-related security methods (`.has_access()` and `.get_accessible_dags()`). Rather than checking the `AUTH_ROLE_PUBLIC` value to check for role permissions, the methods reject access to view or edit any DAGs.

Changes in this PR:
Rather than hardcoding permission rules inside the security methods, this change checks the `AUTH_ROLE_PUBLIC` value and gives anonymous users all permissions linked to the designated role.

**This places security in the hands of the Airflow users. If the value is set to `Admin`, anonymous users will have full admin functionality.**

This also changes how the `Public` role is created. Currently, the `Public` role is created automatically by Flask App Builder. This PR explicitly declares `Public` as a default role with no permissions in `security.py`. This change makes it easier to test.

closes: #13340
(cherry picked from commit 78aa921)

airflow/www/security.py

@@ -140,6 +140,7 @@ class AirflowSecurityManager(SecurityManager, LoggingMixin):  # pylint: disable=
     ###########################################################################
 
     ROLE_CONFIGS = [
+        {'role': 'Public', 'perms': []},
         {'role': 'Viewer', 'perms': VIEWER_PERMISSIONS},
         {
             'role': 'User',
@@ -254,22 +255,24 @@ def get_accessible_dag_ids(self, user) -> Set[str]:
 
     @provide_session
     def get_accessible_dags(self, user_actions, user, session=None):
-        """Generic function to get readable or writable DAGs for authenticated user."""
+        """Generic function to get readable or writable DAGs for user."""
         if user.is_anonymous:
-            return set()
-
-        user_query = (
-            session.query(User)
-            .options(
-                joinedload(User.roles)
-                .subqueryload(Role.permissions)
-                .options(joinedload(PermissionView.permission), joinedload(PermissionView.view_menu))
+            roles = self.get_user_roles(user)
+        else:
+            user_query = (
+                session.query(User)
+                .options(
+                    joinedload(User.roles)
+                    .subqueryload(Role.permissions)
+                    .options(joinedload(PermissionView.permission), joinedload(PermissionView.view_menu))
+                )
+                .filter(User.id == user.id)
+                .first()
             )
-            .filter(User.id == user.id)
-            .first()
-        )
+            roles = user_query.roles
+
         resources = set()
-        for role in user_query.roles:
+        for role in roles:
             for permission in role.permissions:
                 action = permission.permission.name
                 if action not in user_actions:
@@ -348,7 +351,7 @@ def has_access(self, permission, resource, user=None) -> bool:
             user = g.user
 
         if user.is_anonymous:
-            return self.is_item_public(permission, resource)
+            user.roles = self.get_user_roles(user)
 
         has_access = self._has_view_access(user, permission, resource)
         # FAB built-in view access method. Won't work for AllDag access.

tests/www/test_security.py

@@ -125,6 +125,12 @@ def _has_dag_perm(self, perm, dag_id, user):
         #     user = self.user
         return self.security_manager.has_access(perm, self.security_manager.prefixed_dag_id(dag_id), user)
 
+    def _create_dag(self, dag_id):
+        dag_model = DagModel(dag_id=dag_id)
+        self.session.add(dag_model)
+        self.session.commit()
+        self.security_manager.sync_perm_for_dag(dag_id, access_control=None)
+
     def tearDown(self):
         clear_db_runs()
         clear_db_dags()
@@ -170,10 +176,74 @@ def test_update_and_verify_permission_role(self):
         assert role_perms_len == new_role_perms_len
 
     def test_verify_public_role_has_no_permissions(self):
+        public = self.appbuilder.sm.find_role("Public")
+
+        assert public.permissions == []
+
+    def test_verify_default_anon_user_has_no_accessible_dag_ids(self):
         with self.app.app_context():
-            public = self.appbuilder.sm.find_role("Public")
+            user = mock.MagicMock()
+            user.is_anonymous = True
+            self.app.config['AUTH_ROLE_PUBLIC'] = 'Public'
+            assert self.app.appbuilder.sm.get_user_roles(user) == [self.app.appbuilder.sm.get_public_role()]
+
+            self._create_dag("test_dag_id")
+            self.security_manager.sync_roles()
+
+            assert self.security_manager.get_accessible_dag_ids(user) == set()
+
+    def test_verify_default_anon_user_has_no_access_to_specific_dag(self):
+        with self.app.app_context():
+            user = mock.MagicMock()
+            user.is_anonymous = True
+            self.app.config['AUTH_ROLE_PUBLIC'] = 'Public'
+            assert self.app.appbuilder.sm.get_user_roles(user) == [self.app.appbuilder.sm.get_public_role()]
+
+            dag_id = "test_dag_id"
+            self._create_dag(dag_id)
+            self.app.appbuilder.sm.sync_roles()
+
+            assert self.app.appbuilder.sm.can_read_dag(dag_id, user) is False
+            assert self.app.appbuilder.sm.can_edit_dag(dag_id, user) is False
+            assert self._has_dag_perm(permissions.ACTION_CAN_READ, dag_id, user) is False
+            assert self._has_dag_perm(permissions.ACTION_CAN_EDIT, dag_id, user) is False
+
+    def test_verify_anon_user_with_admin_role_has_all_dag_access(self):
+        with self.app.app_context():
+            self.app.config['AUTH_ROLE_PUBLIC'] = 'Admin'
+            user = mock.MagicMock()
+            user.is_anonymous = True
+
+            assert self.app.appbuilder.sm.get_user_roles(user) == [self.app.appbuilder.sm.get_public_role()]
+
+            test_dag_ids = ["test_dag_id_1", "test_dag_id_2", "test_dag_id_3"]
+            for dag_id in test_dag_ids:
+                self._create_dag(dag_id)
+            self.security_manager.sync_roles()
+
+            assert self.security_manager.get_accessible_dag_ids(user) == set(test_dag_ids)
+
+    def test_verify_anon_user_with_admin_role_has_access_to_each_dag(self):
+        with self.app.app_context():
+            user = mock.MagicMock()
+            user.is_anonymous = True
+            self.app.config['AUTH_ROLE_PUBLIC'] = 'Admin'
+
+            # Call `.get_user_roles` bc `user` is a mock and the `user.roles` prop needs to be set.
+            user.roles = self.app.appbuilder.sm.get_user_roles(user)
+            assert user.roles == [self.app.appbuilder.sm.get_public_role()]
+
+            test_dag_ids = ["test_dag_id_1", "test_dag_id_2", "test_dag_id_3"]
+
+            for dag_id in test_dag_ids:
+                self._create_dag(dag_id)
+            self.security_manager.sync_roles()
 
-            assert public.permissions == []
+            for dag_id in test_dag_ids:
+                assert self.app.appbuilder.sm.can_read_dag(dag_id, user) is True
+                assert self.app.appbuilder.sm.can_edit_dag(dag_id, user) is True
+                assert self._has_dag_perm(permissions.ACTION_CAN_READ, dag_id, user) is True
+                assert self._has_dag_perm(permissions.ACTION_CAN_EDIT, dag_id, user) is True
 
     def test_get_user_roles(self):
         user = mock.MagicMock()