Merge pull request #785 from dod-ccpo/check-user-is-in-app-before-adding-env-role

Check user is in app before adding env role

atst/domain/application_roles.py

@@ -6,6 +6,7 @@ from atst.domain.permission_sets import PermissionSets
 class ApplicationRoles(object):
     @classmethod
     def _permission_sets_for_names(cls, set_names):
+        set_names = set(set_names).union({PermissionSets.VIEW_APPLICATION})
         return PermissionSets.get_many(set_names)
 
     @classmethod

atst/domain/exceptions.py

@@ -1,6 +1,7 @@
 class NotFoundError(Exception):
-    def __init__(self, resource_name):
+    def __init__(self, resource_name, resource_id=None):
         self.resource_name = resource_name
+        self.resource_id = resource_id
 
     @property
     def message(self):

atst/models/application.py

@@ -9,7 +9,6 @@ from atst.models.application_role import (
     ApplicationRole,
     Status as ApplicationRoleStatuses,
 )
-
 from atst.database import db
 
 

atst/routes/applications/settings.py

@@ -7,6 +7,8 @@ from atst.domain.applications import Applications
 from atst.forms.app_settings import EnvironmentRolesForm
 from atst.forms.application import ApplicationForm, EditEnvironmentForm
 from atst.domain.authz.decorator import user_can_access_decorator as user_can
+from atst.domain.exceptions import NotFoundError
+
 from atst.models.permissions import Permissions
 from atst.utils.flash import formatted_flash as flash
 
@@ -46,6 +48,14 @@ def serialize_env_member_form_data(application):
     return environments_list
 
 
+def check_users_are_in_application(user_ids, application):
+    existing_ids = [str(role.user_id) for role in application.roles]
+    for user_id in user_ids:
+        if not user_id in existing_ids:
+            raise NotFoundError("application user", user_id)
+    return True
+
+
 @applications_bp.route("/applications/<application_id>/settings")
 @user_can(Permissions.VIEW_APPLICATION, message="view application edit form")
 def settings(application_id):
@@ -128,6 +138,19 @@ def update_env_roles(environment_id):
     env_roles_form = EnvironmentRolesForm(http_request.form)
 
     if env_roles_form.validate():
+
+        try:
+            user_ids = [user["user_id"] for user in env_roles_form.data["team_roles"]]
+            check_users_are_in_application(user_ids, application)
+        except NotFoundError as err:
+            app.logger.warning(
+                "User {} requested environment role change for unauthorized user {} in application {}.".format(
+                    g.current_user.id, err.resource_id, application.id
+                ),
+                extra={"tags": ["update", "failure"], "security_warning": True},
+            )
+
+            raise (err)
         env_data = env_roles_form.data
         Environments.update_env_roles_by_environment(
             environment_id=environment_id, team_roles=env_data["team_roles"]

tests/domain/test_application_roles.py

@@ -14,7 +14,7 @@ def test_create_application_role():
     )
 
     assert application_role.permission_sets == PermissionSets.get_many(
-        [PermissionSets.EDIT_APPLICATION_TEAM]
+        [PermissionSets.EDIT_APPLICATION_TEAM, PermissionSets.VIEW_APPLICATION]
     )
     assert application_role.application == application
     assert application_role.user == user

tests/domain/test_environments.py

@@ -11,6 +11,7 @@ from tests.factories import (
     PortfolioFactory,
     EnvironmentFactory,
     EnvironmentRoleFactory,
+    ApplicationRoleFactory,
 )
 
 
@@ -24,6 +25,9 @@ def test_create_environments():
 def test_update_env_role():
     env_role = EnvironmentRoleFactory.create(role=CSPRole.BASIC_ACCESS.value)
     new_role = CSPRole.TECHNICAL_READ.value
+    ApplicationRoleFactory.create(
+        user=env_role.user, application=env_role.environment.application
+    )
 
     assert Environments.update_env_role(env_role.environment, env_role.user, new_role)
     assert env_role.role == new_role
@@ -31,6 +35,9 @@ def test_update_env_role():
 
 def test_update_env_role_no_access():
     env_role = EnvironmentRoleFactory.create(role=CSPRole.BASIC_ACCESS.value)
+    ApplicationRoleFactory.create(
+        user=env_role.user, application=env_role.environment.application
+    )
 
     assert Environments.update_env_role(env_role.environment, env_role.user, None)
     assert not EnvironmentRoles.get(env_role.user.id, env_role.environment.id)
@@ -48,6 +55,7 @@ def test_update_env_role_no_change():
 def test_update_env_role_creates_cloud_id_for_new_member(session):
     user = UserFactory.create()
     env = EnvironmentFactory.create()
+    ApplicationRoleFactory.create(user=user, application=env.application)
     assert not user.cloud_id
     assert Environments.update_env_role(env, user, CSPRole.TECHNICAL_READ.value)
     assert EnvironmentRoles.get(user.id, env.id)
@@ -65,6 +73,8 @@ def test_update_env_roles_by_environment():
     env_role_3 = EnvironmentRoleFactory.create(
         environment=environment, role=CSPRole.TECHNICAL_READ.value
     )
+    for user in [env_role_1.user, env_role_2.user, env_role_3.user]:
+        ApplicationRoleFactory.create(user=user, application=environment.application)
 
     team_roles = [
         {

tests/models/test_application.py

@@ -4,6 +4,7 @@ from tests.factories import (
     ApplicationFactory,
     ApplicationRoleFactory,
     EnvironmentFactory,
+    UserFactory,
 )
 
 

tests/models/test_environments.py

@@ -1,4 +1,5 @@
 from atst.models import AuditEvent
+from atst.models.environment_role import CSPRole
 from atst.domain.environments import Environments
 from atst.domain.applications import Applications
 
@@ -7,6 +8,7 @@ from tests.factories import (
     UserFactory,
     EnvironmentFactory,
     ApplicationFactory,
+    ApplicationRoleFactory,
 )
 
 
@@ -20,7 +22,10 @@ def test_add_user_to_environment():
     )
     dev_environment = application.environments[0]
 
-    dev_environment = Environments.add_member(dev_environment, developer, "developer")
+    ApplicationRoleFactory.create(user=developer, application=application)
+    dev_environment = Environments.add_member(
+        dev_environment, developer, CSPRole.BASIC_ACCESS.value
+    )
     assert developer in dev_environment.users
 
 

tests/models/test_portfolio_role.py

@@ -10,6 +10,7 @@ from atst.models.portfolio_role import Status
 from atst.models.invitation import Status as InvitationStatus
 from atst.models.audit_event import AuditEvent
 from atst.models.portfolio_role import Status as PortfolioRoleStatus
+from atst.models.environment_role import CSPRole
 from tests.factories import (
     UserFactory,
     InvitationFactory,
@@ -17,6 +18,7 @@ from tests.factories import (
     EnvironmentFactory,
     EnvironmentRoleFactory,
     ApplicationFactory,
+    ApplicationRoleFactory,
     PortfolioFactory,
 )
 from atst.domain.portfolio_roles import PortfolioRoles
@@ -116,6 +118,7 @@ def test_has_env_role_history(session):
     portfolio = PortfolioFactory.create(owner=owner)
     portfolio_role = PortfolioRoleFactory.create(portfolio=portfolio, user=user)
     application = ApplicationFactory.create(portfolio=portfolio)
+    ApplicationRoleFactory.create(user=user, application=application)
     environment = EnvironmentFactory.create(
         application=application, name="new environment!"
     )
@@ -178,8 +181,9 @@ def test_has_environment_roles():
     application = Applications.create(
         portfolio, "my test application", "It's mine.", ["dev", "staging", "prod"]
     )
+    ApplicationRoleFactory.create(user=portfolio_role.user, application=application)
     Environments.add_member(
-        application.environments[0], portfolio_role.user, "developer"
+        application.environments[0], portfolio_role.user, CSPRole.BASIC_ACCESS.value
     )
     assert portfolio_role.has_environment_roles
 

tests/routes/applications/test_settings.py

@@ -1,3 +1,4 @@
+import pytest
 from flask import url_for, get_flashed_messages
 
 from tests.factories import (
@@ -9,12 +10,15 @@ from tests.factories import (
     ApplicationFactory,
     ApplicationRoleFactory,
 )
+from atst.routes.applications.settings import check_users_are_in_application
 
 from atst.domain.applications import Applications
 from atst.domain.environment_roles import EnvironmentRoles
 from atst.domain.environments import Environments
 from atst.domain.permission_sets import PermissionSets
 from atst.domain.portfolios import Portfolios
+from atst.domain.exceptions import NotFoundError
+
 from atst.models.environment_role import CSPRole
 from atst.models.portfolio_role import Status as PortfolioRoleStatus
 from atst.forms.application import EditEnvironmentForm
@@ -193,6 +197,32 @@ def test_user_without_permission_cannot_update_application(client, user_session)
     assert application.description == "Cool stuff happening here!"
 
 
+def test_check_users_are_in_application_raises_NotFoundError():
+    application = ApplicationFactory.create()
+    app_user_1 = UserFactory.create()
+    app_user_2 = UserFactory.create()
+    for user in [app_user_1, app_user_2]:
+        ApplicationRoleFactory.create(user=user, application=application)
+
+    non_app_user = UserFactory.create()
+    user_ids = [app_user_1.id, app_user_2.id, non_app_user.id]
+    with pytest.raises(NotFoundError):
+        check_users_are_in_application(user_ids, application)
+
+
+def test_check_users_are_in_application():
+    application = ApplicationFactory.create()
+    app_user_1 = UserFactory.create()
+    app_user_2 = UserFactory.create()
+    app_user_3 = UserFactory.create()
+
+    for user in [app_user_1, app_user_2, app_user_3]:
+        ApplicationRoleFactory.create(user=user, application=application)
+
+    user_ids = [str(app_user_1.id), str(app_user_2.id), str(app_user_3.id)]
+    assert check_users_are_in_application(user_ids, application)
+
+
 def test_update_team_env_roles(client, user_session):
     environment = EnvironmentFactory.create()
     application = environment.application
@@ -205,6 +235,9 @@ def test_update_team_env_roles(client, user_session):
     env_role_3 = EnvironmentRoleFactory.create(
         environment=environment, role=CSPRole.BASIC_ACCESS.value
     )
+    for user in [env_role_1.user, env_role_2.user, env_role_3.user]:
+        ApplicationRoleFactory.create(user=user, application=application)
+
     app_role = ApplicationRoleFactory.create(application=application)
     form_data = {
         "env_id": environment.id,

tests/test_access.py

@@ -196,7 +196,9 @@ def test_applications_update_team_env_roles(post_url_assert_status):
             ]
         ),
     )
-    ApplicationRoleFactory.create(user=app_member)
+    ApplicationRoleFactory.create(user=app_member, application=application)
+    ApplicationRoleFactory.create(user=ccpo, application=application)
+    ApplicationRoleFactory.create(user=owner, application=application)
 
     url = url_for("applications.update_env_roles", environment_id=environment.id)
     post_url_assert_status(ccpo, url, 302)