Skip to content

AWS auth manager: implement all is_authorized_* methods (but is_authorized_dag) #35928

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vincbeck merged 1 commit into from
Nov 29, 2023
Merged

AWS auth manager: implement all is_authorized_* methods (but is_authorized_dag) #35928

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion airflow/providers/amazon/aws/auth_manager/avp/entities.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,9 +30,16 @@ class AvpEntities(Enum):

ACTION = "Action"
ROLE = "Role"
VARIABLE = "Variable"
USER = "User"

# Resource types
CONFIGURATION = "Configuration"
CONNECTION = "Connection"
DATASET = "Dataset"
POOL = "Pool"
VARIABLE = "Variable"
VIEW = "View"


def get_entity_type(resource_type: AvpEntities) -> str:
"""
Expand Down
39 changes: 34 additions & 5 deletions airflow/providers/amazon/aws/auth_manager/aws_auth_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,13 @@ def is_authorized_configuration(
details: ConfigurationDetails | None = None,
user: BaseUser | None = None,
) -> bool:
return self.is_logged_in()
config_section = details.section if details else None

@o-nikolas o-nikolas Nov 28, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the section is None then is this basically checking if the user is authorized for all/any config?

@vincbeck vincbeck Nov 29, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is checking if the user is authorized for all config. This is a really good question and this has been bugging me for a while. Does auth_manager.is_authorized_variable("GET") means:

  • Does the user has permission to read all variable
  • Does the user has permission to read any variable

I decided to go with the former because this is actually how Amazon Verified Permissions work. You cannot ask AVP to check whether the user is authorized to access ANY variable (or any type of resource type).

But this has some implications. Example: As a user I am allowed to access only variables starting with "tmp_" (thus not all variables), thus I wont be allowed to see the list of variables (or even the menu). The reason why the user would not be able to see the menu "Variables" or the list of variables is because to display those, we check whether the user has permissions to read all variables. So this is not perfect either (and we might change it) but I guess we can leave that interrogation for later/another PR

@vincbeck vincbeck Nov 29, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But I dont know if this is something we should thing about because Airflow is not ready for such use cases. The views (this is the first time that comes to my mind) are displaying all resources from database without filtering (we could change that of course)

@vincbeck vincbeck Nov 29, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@shubham22. Related to the discussion we had

@o-nikolas o-nikolas Nov 29, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good discussion, it sounds like we'll likely merge this either way so I'll approve since the code looks good otherwise. But please keep the discussion going to see where we net out on this topic.

return self.avp_facade.is_authorized(
method=method,
entity_type=AvpEntities.CONFIGURATION,
user=user or self.get_user(),
entity_id=config_section,
)

def is_authorized_cluster_activity(self, *, method: ResourceMethod, user: BaseUser | None = None) -> bool:
return self.is_logged_in()
Expand All @@ -103,7 +109,13 @@ def is_authorized_connection(
details: ConnectionDetails | None = None,
user: BaseUser | None = None,
) -> bool:
return self.is_logged_in()
connection_id = details.conn_id if details else None
return self.avp_facade.is_authorized(
method=method,
entity_type=AvpEntities.CONNECTION,
user=user or self.get_user(),
entity_id=connection_id,
)

def is_authorized_dag(
self,
Expand All @@ -118,12 +130,24 @@ def is_authorized_dag(
def is_authorized_dataset(
self, *, method: ResourceMethod, details: DatasetDetails | None = None, user: BaseUser | None = None
) -> bool:
return self.is_logged_in()
dataset_uri = details.uri if details else None
return self.avp_facade.is_authorized(
method=method,
entity_type=AvpEntities.DATASET,
user=user or self.get_user(),
entity_id=dataset_uri,
)

def is_authorized_pool(
self, *, method: ResourceMethod, details: PoolDetails | None = None, user: BaseUser | None = None
) -> bool:
return self.is_logged_in()
pool_name = details.name if details else None
return self.avp_facade.is_authorized(
method=method,
entity_type=AvpEntities.POOL,
user=user or self.get_user(),
entity_id=pool_name,
)

def is_authorized_variable(
self, *, method: ResourceMethod, details: VariableDetails | None = None, user: BaseUser | None = None
Expand All @@ -142,7 +166,12 @@ def is_authorized_view(
access_view: AccessView,
user: BaseUser | None = None,
) -> bool:
return self.is_logged_in()
return self.avp_facade.is_authorized(
method="GET",
entity_type=AvpEntities.VIEW,
user=user or self.get_user(),
entity_id=access_view.value,

@o-nikolas o-nikolas Nov 28, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why don't we have a check for this value and default to None like the others?

@vincbeck vincbeck Nov 28, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

access_view is a required field

@o-nikolas o-nikolas Nov 28, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why is this one required but not the others? I think maybe the answer to my other comment might answer this as well.

@vincbeck vincbeck Nov 29, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The other are optional because when you do authorization request, you might not need to specify a resource ID. Example: "Does the user have permission to create a Variable". There is no resource ID here. Here this is not really a resource ID but more a specific of which view the user is trying to access. We must specify which view the user is trying to access

)

def get_url_login(self, **kwargs) -> str:
return url_for("AwsAuthManagerAuthenticationViews.login")
Expand Down
135 changes: 133 additions & 2 deletions tests/providers/amazon/aws/auth_manager/test_aws_auth_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,14 @@
import pytest
from flask import Flask, session

from airflow.auth.managers.models.resource_details import VariableDetails
from airflow.auth.managers.models.resource_details import (
AccessView,
ConfigurationDetails,
ConnectionDetails,
DatasetDetails,
PoolDetails,
VariableDetails,
)
from airflow.providers.amazon.aws.auth_manager.avp.entities import AvpEntities
from airflow.providers.amazon.aws.auth_manager.aws_auth_manager import AwsAuthManager
from airflow.providers.amazon.aws.auth_manager.security_manager.aws_security_manager_override import (
Expand Down Expand Up @@ -110,6 +117,108 @@ def test_is_logged_in_return_false_when_no_user_in_session(self, auth_manager, a

assert result is False

@pytest.mark.parametrize(
"details, user, expected_user, expected_entity_id",
[
(None, None, ANY, None),
(ConfigurationDetails(section="test"), mock, mock, "test"),
],
)
@patch.object(AwsAuthManager, "avp_facade")
@patch.object(AwsAuthManager, "get_user")
def test_is_authorized_configuration(
self, mock_get_user, mock_avp_facade, details, user, expected_user, expected_entity_id, auth_manager
):
is_authorized = Mock()
mock_avp_facade.is_authorized = is_authorized

method: ResourceMethod = "GET"
auth_manager.is_authorized_configuration(method=method, details=details, user=user)

if not user:
mock_get_user.assert_called_once()
is_authorized.assert_called_once_with(
method=method,
entity_type=AvpEntities.CONFIGURATION,
user=expected_user,
entity_id=expected_entity_id,
)

@pytest.mark.parametrize(
"details, user, expected_user, expected_entity_id",
[
(None, None, ANY, None),
(ConnectionDetails(conn_id="conn_id"), mock, mock, "conn_id"),
],
)
@patch.object(AwsAuthManager, "avp_facade")
@patch.object(AwsAuthManager, "get_user")
def test_is_authorized_connection(
self, mock_get_user, mock_avp_facade, details, user, expected_user, expected_entity_id, auth_manager
):
is_authorized = Mock()
mock_avp_facade.is_authorized = is_authorized

method: ResourceMethod = "GET"
auth_manager.is_authorized_connection(method=method, details=details, user=user)

if not user:
mock_get_user.assert_called_once()
is_authorized.assert_called_once_with(
method=method,
entity_type=AvpEntities.CONNECTION,
user=expected_user,
entity_id=expected_entity_id,
)

@pytest.mark.parametrize(
"details, user, expected_user, expected_entity_id",
[
(None, None, ANY, None),
(DatasetDetails(uri="uri"), mock, mock, "uri"),
],
)
@patch.object(AwsAuthManager, "avp_facade")
@patch.object(AwsAuthManager, "get_user")
def test_is_authorized_dataset(
self, mock_get_user, mock_avp_facade, details, user, expected_user, expected_entity_id, auth_manager
):
is_authorized = Mock()
mock_avp_facade.is_authorized = is_authorized

method: ResourceMethod = "GET"
auth_manager.is_authorized_dataset(method=method, details=details, user=user)

if not user:
mock_get_user.assert_called_once()
is_authorized.assert_called_once_with(
method=method, entity_type=AvpEntities.DATASET, user=expected_user, entity_id=expected_entity_id
)

@pytest.mark.parametrize(
"details, user, expected_user, expected_entity_id",
[
(None, None, ANY, None),
(PoolDetails(name="pool1"), mock, mock, "pool1"),
],
)
@patch.object(AwsAuthManager, "avp_facade")
@patch.object(AwsAuthManager, "get_user")
def test_is_authorized_pool(
self, mock_get_user, mock_avp_facade, details, user, expected_user, expected_entity_id, auth_manager
):
is_authorized = Mock()
mock_avp_facade.is_authorized = is_authorized

method: ResourceMethod = "GET"
auth_manager.is_authorized_pool(method=method, details=details, user=user)

if not user:
mock_get_user.assert_called_once()
is_authorized.assert_called_once_with(
method=method, entity_type=AvpEntities.POOL, user=expected_user, entity_id=expected_entity_id
)

@pytest.mark.parametrize(
"details, user, expected_user, expected_entity_id",
[
Expand All @@ -126,7 +235,6 @@ def test_is_authorized_variable(
mock_avp_facade.is_authorized = is_authorized

method: ResourceMethod = "GET"

auth_manager.is_authorized_variable(method=method, details=details, user=user)

if not user:
Expand All @@ -135,6 +243,29 @@ def test_is_authorized_variable(
method=method, entity_type=AvpEntities.VARIABLE, user=expected_user, entity_id=expected_entity_id
)

@pytest.mark.parametrize(
"access_view, user, expected_user",
[
(AccessView.CLUSTER_ACTIVITY, None, ANY),
(AccessView.PLUGINS, mock, mock),
],
)
@patch.object(AwsAuthManager, "avp_facade")
@patch.object(AwsAuthManager, "get_user")
def test_is_authorized_view(
self, mock_get_user, mock_avp_facade, access_view, user, expected_user, auth_manager
):
is_authorized = Mock()
mock_avp_facade.is_authorized = is_authorized

auth_manager.is_authorized_view(access_view=access_view, user=user)

if not user:
mock_get_user.assert_called_once()
is_authorized.assert_called_once_with(
method="GET", entity_type=AvpEntities.VIEW, user=expected_user, entity_id=access_view.value
)

@patch("airflow.providers.amazon.aws.auth_manager.aws_auth_manager.url_for")
def test_get_url_login(self, mock_url_for, auth_manager):
auth_manager.get_url_login()
Expand Down