mirror of
https://github.com/hanzoai/sentinel.git
synced 2026-09-29 22:34:33 +00:00
957 lines
43 KiB
Python
957 lines
43 KiB
Python
from datetime import timedelta
|
|
from unittest.mock import Mock, patch
|
|
|
|
from django.contrib.auth.models import AnonymousUser
|
|
from django.test import override_settings
|
|
from django.utils import timezone
|
|
|
|
from sentry.auth import access
|
|
from sentry.auth.access import Access, NoAccess
|
|
from sentry.auth.providers.dummy import DummyProvider
|
|
from sentry.auth.superuser import SUPERUSER_READONLY_SCOPES, SUPERUSER_SCOPES
|
|
from sentry.constants import ObjectStatus
|
|
from sentry.models.apikey import ApiKey
|
|
from sentry.models.authidentity import AuthIdentity
|
|
from sentry.models.authprovider import AuthProvider
|
|
from sentry.models.organization import Organization
|
|
from sentry.models.team import TeamStatus
|
|
from sentry.models.user import User
|
|
from sentry.models.userrole import UserRole
|
|
from sentry.services.hybrid_cloud.access.service import access_service
|
|
from sentry.services.hybrid_cloud.organization import organization_service
|
|
from sentry.silo import SiloMode
|
|
from sentry.testutils.cases import TestCase
|
|
from sentry.testutils.helpers import with_feature
|
|
from sentry.testutils.silo import all_silo_test, assume_test_silo_mode, no_silo_test
|
|
|
|
|
|
def silo_from_user(
|
|
user,
|
|
organization=None,
|
|
scopes=None,
|
|
is_superuser=False,
|
|
is_staff=False,
|
|
) -> Access:
|
|
rpc_user_org_context = None
|
|
if organization:
|
|
rpc_user_org_context = organization_service.get_organization_by_id(
|
|
id=organization.id, user_id=user.id
|
|
)
|
|
return access.from_user_and_rpc_user_org_context(
|
|
user=user,
|
|
rpc_user_org_context=rpc_user_org_context,
|
|
is_superuser=is_superuser,
|
|
is_staff=is_staff,
|
|
scopes=scopes,
|
|
)
|
|
|
|
|
|
def silo_from_request(request, organization: Organization | None = None, scopes=None) -> Access:
|
|
rpc_user_org_context = None
|
|
if organization:
|
|
rpc_user_org_context = organization_service.get_organization_by_id(
|
|
id=organization.id, user_id=request.user.id
|
|
)
|
|
return access.from_request_org_and_scopes(
|
|
request=request, rpc_user_org_context=rpc_user_org_context, scopes=scopes
|
|
)
|
|
|
|
|
|
class AccessFactoryTestCase(TestCase):
|
|
def from_user(self, *args, **kwds):
|
|
if SiloMode.get_current_mode() == SiloMode.MONOLITH:
|
|
return access.from_user(*args, **kwds)
|
|
return silo_from_user(*args, **kwds)
|
|
|
|
def from_request(self, *args, **kwds):
|
|
if SiloMode.get_current_mode() == SiloMode.MONOLITH:
|
|
return access.from_request(*args, **kwds)
|
|
return silo_from_request(*args, **kwds)
|
|
|
|
@assume_test_silo_mode(SiloMode.CONTROL)
|
|
def create_api_key(self, organization: Organization, **kwds):
|
|
return ApiKey.objects.create(organization_id=organization.id, **kwds)
|
|
|
|
@assume_test_silo_mode(SiloMode.CONTROL)
|
|
def create_auth_provider(self, organization: Organization, **kwds):
|
|
return AuthProvider.objects.create(organization_id=organization.id, **kwds)
|
|
|
|
@assume_test_silo_mode(SiloMode.CONTROL)
|
|
def create_auth_identity(self, auth_provider: AuthProvider, user: User, **kwds):
|
|
return AuthIdentity.objects.create(auth_provider=auth_provider, user=user, **kwds)
|
|
|
|
|
|
@all_silo_test
|
|
class FromUserTest(AccessFactoryTestCase):
|
|
def test_no_access(self):
|
|
organization = self.create_organization()
|
|
team = self.create_team(organization=organization)
|
|
project = self.create_project(organization=organization, teams=[team])
|
|
user = self.create_user()
|
|
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert not result.sso_is_valid
|
|
assert not result.requires_sso
|
|
assert not result.scopes
|
|
assert not result.has_team_access(team)
|
|
assert not result.has_team_scope(team, "project:read")
|
|
assert not result.has_project_access(project)
|
|
assert not result.has_projects_access([project])
|
|
assert not result.has_project_scope(project, "project:read")
|
|
assert not result.has_project_membership(project)
|
|
assert not result.permissions
|
|
|
|
def test_no_deleted_projects(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization(owner=self.user)
|
|
|
|
team = self.create_team(organization=organization)
|
|
self.create_member(organization=organization, user=user, role="owner", teams=[team])
|
|
deleted_project = self.create_project(
|
|
organization=organization, status=ObjectStatus.PENDING_DELETION, teams=[team]
|
|
)
|
|
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert result.has_project_access(deleted_project) is False
|
|
assert result.has_project_membership(deleted_project) is False
|
|
assert len(result.project_ids_with_team_membership) == 0
|
|
|
|
def test_no_deleted_teams(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization(owner=self.user)
|
|
|
|
team = self.create_team(organization=organization)
|
|
deleted_team = self.create_team(
|
|
organization=organization, status=TeamStatus.PENDING_DELETION
|
|
)
|
|
self.create_member(
|
|
organization=organization, user=user, role="owner", teams=[team, deleted_team]
|
|
)
|
|
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert result.has_team_access(team) is True
|
|
assert result.has_team_access(deleted_team) is False
|
|
assert result.team_ids_with_membership == frozenset({team.id})
|
|
|
|
def test_unique_projects(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization(owner=self.user)
|
|
|
|
team = self.create_team(organization=organization)
|
|
other_team = self.create_team(organization=organization)
|
|
self.create_member(
|
|
organization=organization, user=user, role="owner", teams=[team, other_team]
|
|
)
|
|
project = self.create_project(organization=organization, teams=[team, other_team])
|
|
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert result.has_project_access(project)
|
|
assert len(result.project_ids_with_team_membership) == 1
|
|
|
|
def test_mixed_access(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization(flags=0) # disable default allow_joinleave
|
|
team = self.create_team(organization=organization)
|
|
team_no_access = self.create_team(organization=organization)
|
|
project = self.create_project(organization=organization, teams=[team])
|
|
project_no_access = self.create_project(organization=organization, teams=[team_no_access])
|
|
self.create_member(organization=organization, user=user, teams=[team])
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert result.has_project_access(project)
|
|
assert not result.has_project_access(project_no_access)
|
|
assert not result.has_projects_access([project, project_no_access])
|
|
|
|
def test_owner_all_teams(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization(owner=self.user)
|
|
member = self.create_member(organization=organization, user=user, role="owner")
|
|
team = self.create_team(organization=organization)
|
|
project = self.create_project(organization=organization, teams=[team])
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert result.sso_is_valid
|
|
assert not result.requires_sso
|
|
assert result.scopes == member.get_scopes()
|
|
assert result.has_team_access(team)
|
|
assert result.has_team_scope(team, "project:read")
|
|
assert result.has_project_access(project)
|
|
assert result.has_projects_access([project])
|
|
assert result.has_project_scope(project, "project:read")
|
|
assert result.has_any_project_scope(project, ["project:read", "project:admin"])
|
|
|
|
# owners should have access but not membership
|
|
assert result.has_project_membership(project) is False
|
|
|
|
def test_member_no_teams_closed_membership(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization(
|
|
owner=self.user, flags=0 # disable default allow_joinleave
|
|
)
|
|
member = self.create_member(organization=organization, user=user, role="member")
|
|
team = self.create_team(organization=organization)
|
|
project = self.create_project(organization=organization, teams=[team])
|
|
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert result.sso_is_valid
|
|
assert not result.requires_sso
|
|
assert result.scopes == member.get_scopes()
|
|
assert not result.has_team_access(team)
|
|
assert not result.has_team_scope(team, "project:read")
|
|
assert not result.has_project_access(project)
|
|
assert not result.has_projects_access([project])
|
|
assert not result.has_project_scope(project, "project:read")
|
|
assert not result.has_any_project_scope(project, ["project:read", "project:admin"])
|
|
assert not result.has_project_membership(project)
|
|
|
|
def test_member_no_teams_open_membership(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization(
|
|
owner=self.user, flags=Organization.flags.allow_joinleave
|
|
)
|
|
member = self.create_member(organization=organization, user=user, role="member", teams=())
|
|
team = self.create_team(organization=organization)
|
|
project = self.create_project(organization=organization, teams=[team])
|
|
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert result.sso_is_valid
|
|
assert not result.requires_sso
|
|
assert result.scopes == member.get_scopes()
|
|
assert result.has_team_access(team)
|
|
assert result.has_team_scope(team, "project:read")
|
|
assert result.has_project_access(project)
|
|
assert result.has_projects_access([project])
|
|
assert result.has_project_scope(project, "project:read")
|
|
assert not result.has_project_scope(project, "project:write")
|
|
assert result.has_any_project_scope(project, ["project:read", "project:write"])
|
|
assert not result.has_any_project_scope(project, ["project:write", "project:admin"])
|
|
assert not result.has_project_membership(project)
|
|
|
|
def test_team_restricted_org_member_access(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization()
|
|
team = self.create_team(organization=organization)
|
|
project = self.create_project(organization=organization, teams=[team])
|
|
member = self.create_member(organization=organization, user=user, teams=[team])
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert result.sso_is_valid
|
|
assert not result.requires_sso
|
|
assert result.scopes == member.get_scopes()
|
|
assert result.has_team_access(team)
|
|
assert result.has_team_scope(team, "project:read")
|
|
assert result.has_project_access(project)
|
|
assert result.has_projects_access([project])
|
|
assert result.has_project_scope(project, "project:read")
|
|
assert not result.has_project_scope(project, "project:write")
|
|
assert result.has_any_project_scope(project, ["project:read", "project:write"])
|
|
assert not result.has_any_project_scope(project, ["project:write", "project:admin"])
|
|
assert result.has_project_membership(project)
|
|
|
|
@with_feature("organizations:team-roles")
|
|
def test_has_project_scope_from_team_role(self):
|
|
organization = self.create_organization()
|
|
team = self.create_team(organization=organization)
|
|
project = self.create_project(organization=organization, teams=[team])
|
|
team_other = self.create_team(organization=organization)
|
|
project_other = self.create_project(organization=organization, teams=[team_other])
|
|
|
|
# Team Contributor
|
|
user = self.create_user()
|
|
member = self.create_member(organization=organization, user=user)
|
|
self.create_team_membership(team, member, role="contributor")
|
|
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
# Does not have scopes from org-role
|
|
assert not result.has_scope("team:admin")
|
|
assert not result.has_scope("team:write")
|
|
assert not result.has_scope("project:admin")
|
|
assert not result.has_scope("project:write")
|
|
|
|
# Has scopes from team-role
|
|
assert not result.has_team_scope(team, "team:admin")
|
|
assert not result.has_team_scope(team, "team:write")
|
|
assert result.has_team_scope(team, "team:read")
|
|
assert not result.has_project_scope(project, "project:admin")
|
|
assert not result.has_project_scope(project, "project:write")
|
|
assert result.has_project_scope(project, "project:read")
|
|
|
|
# Team Admin
|
|
user = self.create_user()
|
|
member = self.create_member(organization=organization, user=user)
|
|
self.create_team_membership(team, member, role="admin")
|
|
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
for result in results:
|
|
# Does not have scopes from org-role
|
|
assert not result.has_scope("team:admin")
|
|
assert not result.has_scope("team:write")
|
|
assert not result.has_scope("project:admin")
|
|
assert not result.has_scope("project:write")
|
|
|
|
# Has scopes from team-role
|
|
assert result.has_team_scope(team, "team:admin")
|
|
assert result.has_team_scope(team, "team:write")
|
|
assert result.has_team_scope(team, "team:read")
|
|
assert result.has_project_scope(project, "project:admin")
|
|
assert result.has_project_scope(project, "project:write")
|
|
assert result.has_project_scope(project, "project:read")
|
|
|
|
# Does not have scope from other team
|
|
assert not result.has_team_scope(team_other, "team:admin")
|
|
assert not result.has_team_scope(team_other, "team:write")
|
|
assert result.has_team_scope(team_other, "team:read")
|
|
assert not result.has_project_scope(project_other, "project:admin")
|
|
assert not result.has_project_scope(project_other, "project:write")
|
|
assert result.has_project_scope(project_other, "project:read")
|
|
|
|
def test_get_organization_roles_from_teams(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization()
|
|
owner_team = self.create_team(organization=organization, org_role="owner")
|
|
admin_team = self.create_team(organization=organization, org_role="admin")
|
|
self.create_member(organization=organization, user=user, teams=[owner_team, admin_team])
|
|
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert "owner" in result.roles
|
|
assert "member" in result.roles
|
|
assert "admin" in result.roles
|
|
|
|
def test_unlinked_sso(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization(owner=user)
|
|
self.create_team(organization=organization)
|
|
ap = self.create_auth_provider(organization=organization, provider="dummy")
|
|
self.create_auth_identity(auth_provider=ap, user=user)
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert not result.sso_is_valid
|
|
assert result.requires_sso
|
|
|
|
def test_last_verified_sso(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization(owner=user)
|
|
ap = self.create_auth_provider(organization=organization, provider="dummy")
|
|
ai = self.create_auth_identity(auth_provider=ap, user=user)
|
|
|
|
om = organization_service.check_membership_by_id(
|
|
organization_id=organization.id, user_id=ai.user_id
|
|
)
|
|
assert om
|
|
setattr(om.flags, "sso:linked", True)
|
|
organization_service.update_membership_flags(organization_member=om)
|
|
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert result.sso_is_valid
|
|
assert result.requires_sso
|
|
|
|
# If the auth identity has not been updated in awhile, it is not valid.
|
|
with assume_test_silo_mode(SiloMode.CONTROL):
|
|
ai.update(last_verified=timezone.now() - timedelta(days=10))
|
|
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
for result in results:
|
|
assert not result.sso_is_valid
|
|
assert result.requires_sso
|
|
|
|
# but it is valid if the requires_fresh is False
|
|
with patch.object(DummyProvider, "requires_refresh", False):
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
for result in results:
|
|
assert result.sso_is_valid
|
|
assert result.requires_sso
|
|
|
|
def test_unlinked_sso_with_owner_from_team(self):
|
|
organization = self.create_organization()
|
|
ap = self.create_auth_provider(organization=organization, provider="dummy")
|
|
user = self.create_user()
|
|
owner_team = self.create_team(organization=organization)
|
|
self.create_member(organization=organization, user=user, teams=[owner_team])
|
|
self.create_auth_identity(auth_provider=ap, user=user)
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert not result.sso_is_valid
|
|
assert result.requires_sso
|
|
|
|
def test_unlinked_sso_with_no_owners(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization(owner=user)
|
|
self.create_team(organization=organization)
|
|
self.create_auth_provider(organization=organization, provider="dummy")
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert not result.sso_is_valid
|
|
assert not result.requires_sso
|
|
|
|
def test_sso_without_link_requirement(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization(owner=user)
|
|
self.create_team(organization=organization)
|
|
self.create_auth_provider(
|
|
organization=organization, provider="dummy", flags=AuthProvider.flags.allow_unlinked
|
|
)
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert result.sso_is_valid
|
|
assert not result.requires_sso
|
|
|
|
def test_anonymous_user(self):
|
|
user = self.create_user()
|
|
anon_user = AnonymousUser()
|
|
organization = self.create_organization(owner=user)
|
|
# TODO: make test work with from_request
|
|
result = self.from_user(anon_user, organization)
|
|
assert result is access.DEFAULT
|
|
|
|
def test_inactive_user(self):
|
|
user = self.create_user(is_active=False)
|
|
organization = self.create_organization(owner=user)
|
|
request = self.make_request(user=user)
|
|
results = [self.from_user(user, organization), self.from_request(request, organization)]
|
|
|
|
for result in results:
|
|
assert result is access.DEFAULT
|
|
|
|
def test_user_permissions_as_superuser(self):
|
|
user = self.create_user(is_superuser=True)
|
|
self.add_user_permission(user, "test.permission")
|
|
|
|
result = self.from_user(user)
|
|
assert not result.has_permission("test.permission")
|
|
|
|
result = self.from_user(user, is_superuser=True)
|
|
assert result.has_permission("test.permission")
|
|
|
|
def test_user_permissions_as_staff(self):
|
|
user = self.create_user(is_staff=True)
|
|
self.add_user_permission(user, "test.permission")
|
|
|
|
result = self.from_user(user)
|
|
assert not result.has_permission("test.permission")
|
|
|
|
result = self.from_user(user, is_staff=True)
|
|
assert result.has_permission("test.permission")
|
|
|
|
@with_feature("organizations:team-roles")
|
|
def test_enforce_upper_bound_scope(self):
|
|
organization = self.create_organization()
|
|
team = self.create_team(organization=organization)
|
|
project = self.create_project(organization=organization, teams=[team])
|
|
team_other = self.create_team(organization=organization)
|
|
project_other = self.create_project(organization=organization, teams=[team_other])
|
|
|
|
# Team Admin
|
|
user = self.create_user()
|
|
member = self.create_member(organization=organization, user=user)
|
|
self.create_team_membership(team, member, role="admin")
|
|
|
|
request = self.make_request(user=user)
|
|
|
|
results = [
|
|
self.from_user(user, organization, scopes=["org:read", "team:admin"]),
|
|
self.from_request(request, organization, scopes=["org:read", "team:admin"]),
|
|
]
|
|
for result in results:
|
|
# Does not have scopes from org-role
|
|
assert not result.has_scope("org:admin")
|
|
assert not result.has_scope("org:write")
|
|
assert result.has_scope("org:read")
|
|
assert not result.has_scope("team:admin") # Org-member do not have team:admin scope
|
|
assert not result.has_scope("team:read")
|
|
assert not result.has_scope("team:write")
|
|
assert not result.has_scope("project:admin")
|
|
assert not result.has_scope("project:write")
|
|
assert not result.has_scope("project:read")
|
|
|
|
# Has scopes from team-role
|
|
assert result.has_team_scope(team, "team:admin") # From being a team-admin
|
|
assert not result.has_team_scope(team, "team:write")
|
|
assert not result.has_team_scope(team, "team:read")
|
|
assert not result.has_project_scope(project, "project:admin")
|
|
assert not result.has_project_scope(project, "project:write")
|
|
assert not result.has_project_scope(project, "project:read")
|
|
|
|
# Does not have scope from other team
|
|
assert not result.has_team_scope(team_other, "team:admin")
|
|
assert not result.has_team_scope(team_other, "team:write")
|
|
assert not result.has_team_scope(team_other, "team:read")
|
|
assert not result.has_project_scope(project_other, "project:admin")
|
|
assert not result.has_project_scope(project_other, "project:write")
|
|
assert not result.has_project_scope(project_other, "project:read")
|
|
|
|
|
|
@all_silo_test
|
|
class FromRequestTest(AccessFactoryTestCase):
|
|
def setUp(self) -> None:
|
|
self.superuser = self.create_user(is_superuser=True)
|
|
self.add_user_permission(self.superuser, "test.permission")
|
|
self.staff = self.create_user(is_staff=True)
|
|
self.add_user_permission(self.staff, "test.permission")
|
|
|
|
self.org = self.create_organization()
|
|
with assume_test_silo_mode(SiloMode.CONTROL):
|
|
AuthProvider.objects.create(organization_id=self.org.id)
|
|
|
|
self.team1 = self.create_team(organization=self.org)
|
|
self.project1 = self.create_project(organization=self.org, teams=[self.team1])
|
|
self.team2 = self.create_team(organization=self.org)
|
|
self.project2 = self.create_project(organization=self.org, teams=[self.team2])
|
|
|
|
super().setUp()
|
|
|
|
def _assert_memberships(self, result: Access) -> None:
|
|
assert result.role == "admin"
|
|
|
|
assert result.team_ids_with_membership == frozenset({self.team1.id})
|
|
assert result.has_team_access(self.team1)
|
|
assert result.project_ids_with_team_membership == frozenset({self.project1.id})
|
|
assert result.has_project_access(self.project1)
|
|
assert result.has_project_membership(self.project1)
|
|
assert not result.has_project_membership(self.project2)
|
|
|
|
# Even if not superuser/staff, still has these because of role.is_global
|
|
# which checks that open membership is on
|
|
assert result.has_global_access
|
|
assert result.has_team_access(self.team2)
|
|
assert result.has_project_access(self.project2)
|
|
|
|
def test_superuser_user_permissions(self):
|
|
request = self.make_request(user=self.superuser, is_superuser=False)
|
|
result = self.from_request(request)
|
|
assert not result.has_permission("test.permission")
|
|
|
|
request = self.make_request(user=self.superuser, is_superuser=True)
|
|
result = self.from_request(request)
|
|
assert result.has_permission("test.permission")
|
|
|
|
def test_staff_user_permissions(self):
|
|
request = self.make_request(user=self.staff, is_staff=False)
|
|
result = self.from_request(request)
|
|
assert not result.has_permission("test.permission")
|
|
|
|
request = self.make_request(user=self.staff, is_staff=True)
|
|
result = self.from_request(request)
|
|
assert result.has_permission("test.permission")
|
|
|
|
def test_superuser_scopes(self):
|
|
# superuser not in organization
|
|
request = self.make_request(user=self.superuser, is_superuser=True)
|
|
|
|
# needs org in request in order to assign any scopes
|
|
result = self.from_request(request, self.org)
|
|
assert result.scopes == SUPERUSER_SCOPES
|
|
|
|
# superuser in organization
|
|
self.create_member(user=self.superuser, organization=self.org, role="member")
|
|
|
|
result = self.from_request(request, self.org)
|
|
assert result.scopes == SUPERUSER_SCOPES
|
|
|
|
@with_feature("auth:enterprise-superuser-read-write")
|
|
@override_settings(SENTRY_SELF_HOSTED=False)
|
|
def test_superuser_readonly_scopes(self):
|
|
# superuser not in organization
|
|
request = self.make_request(user=self.superuser, is_superuser=True)
|
|
|
|
result = self.from_request(request, self.org)
|
|
assert result.scopes == SUPERUSER_READONLY_SCOPES
|
|
|
|
# superuser in organization
|
|
self.create_member(user=self.superuser, organization=self.org, role="member")
|
|
|
|
result = self.from_request(request, self.org)
|
|
assert result.scopes == SUPERUSER_READONLY_SCOPES
|
|
|
|
@with_feature("auth:enterprise-superuser-read-write")
|
|
@override_settings(SENTRY_SELF_HOSTED=False)
|
|
def test_superuser_write_scopes(self):
|
|
self.add_user_permission(self.superuser, "superuser.write")
|
|
|
|
# superuser not in organization
|
|
request = self.make_request(user=self.superuser, is_superuser=True)
|
|
|
|
result = self.from_request(request, self.org)
|
|
assert result.scopes == SUPERUSER_SCOPES
|
|
|
|
# superuser in organization
|
|
self.create_member(user=self.superuser, organization=self.org, role="member")
|
|
|
|
result = self.from_request(request, self.org)
|
|
assert result.scopes == SUPERUSER_SCOPES
|
|
|
|
@with_feature("auth:enterprise-superuser-read-write")
|
|
@override_settings(SENTRY_SELF_HOSTED=False)
|
|
def test_superuser_in_organization_write_scopes(self):
|
|
self.add_user_permission(self.superuser, "superuser.write")
|
|
|
|
request = self.make_request(user=self.superuser, is_superuser=True)
|
|
|
|
result = self.from_request(request, self.org)
|
|
assert result.scopes == SUPERUSER_SCOPES
|
|
|
|
def test_superuser_in_organization(self):
|
|
self.create_member(
|
|
user=self.superuser, organization=self.org, role="admin", teams=[self.team1]
|
|
)
|
|
|
|
request = self.make_request(self.superuser, is_superuser=False)
|
|
result = self.from_request(request, self.org)
|
|
self._assert_memberships(result)
|
|
assert not result.has_permission("test.permission")
|
|
assert "org:superuser" not in result.scopes
|
|
|
|
request = self.make_request(user=self.superuser, is_superuser=True)
|
|
result = self.from_request(request, self.org)
|
|
self._assert_memberships(result)
|
|
assert result.has_permission("test.permission")
|
|
assert result.requires_sso
|
|
assert not result.sso_is_valid
|
|
# org:superuser is only attached when an org is present + active superuser
|
|
assert "org:superuser" in result.scopes
|
|
|
|
def test_staff_in_organization(self):
|
|
self.create_member(user=self.staff, organization=self.org, role="admin", teams=[self.team1])
|
|
|
|
request = self.make_request(self.staff, is_staff=False)
|
|
result = self.from_request(request, self.org)
|
|
self._assert_memberships(result)
|
|
assert not result.has_permission("test.permission")
|
|
|
|
request = self.make_request(user=self.staff, is_staff=True)
|
|
result = self.from_request(request, self.org)
|
|
self._assert_memberships(result)
|
|
assert result.has_permission("test.permission")
|
|
assert result.requires_sso
|
|
assert not result.sso_is_valid
|
|
|
|
def test_superuser_with_organization_without_membership(self):
|
|
request = self.make_request(user=self.superuser, is_superuser=True)
|
|
result = self.from_request(request, self.org)
|
|
assert result.has_permission("test.permission")
|
|
# org:superuser is only attached when an org is present + active superuser
|
|
assert "org:superuser" in result.scopes
|
|
|
|
assert not result.requires_sso
|
|
assert result.sso_is_valid
|
|
assert result.team_ids_with_membership == frozenset()
|
|
assert result.has_team_access(self.team1)
|
|
assert result.project_ids_with_team_membership == frozenset()
|
|
assert result.has_project_access(self.project1)
|
|
|
|
def test_staff_with_organization_without_membership(self):
|
|
request = self.make_request(user=self.staff, is_staff=True)
|
|
result = self.from_request(request, self.org)
|
|
assert result.has_permission("test.permission")
|
|
|
|
assert not result.requires_sso
|
|
# We do not allow staff who are not members of orgs bypass SSO.
|
|
assert not result.sso_is_valid
|
|
# Staff should not have team or project access for another organization
|
|
assert result.team_ids_with_membership == frozenset()
|
|
assert not result.has_team_access(self.team1)
|
|
assert result.project_ids_with_team_membership == frozenset()
|
|
assert not result.has_project_access(self.project1)
|
|
|
|
def test_member_role_in_organization_closed_membership(self):
|
|
# disable default allow_joinleave
|
|
with assume_test_silo_mode(SiloMode.REGION):
|
|
self.org.update(flags=0)
|
|
member_user = self.create_user(is_superuser=False)
|
|
self.create_member(
|
|
user=member_user, organization=self.org, role="member", teams=[self.team1]
|
|
)
|
|
|
|
request = self.make_request(member_user, is_superuser=False)
|
|
result = self.from_request(request, self.org)
|
|
|
|
assert result.role == "member"
|
|
assert result.team_ids_with_membership == frozenset({self.team1.id})
|
|
assert result.has_team_access(self.team1)
|
|
assert result.project_ids_with_team_membership == frozenset({self.project1.id})
|
|
assert result.has_project_access(self.project1)
|
|
assert result.has_project_membership(self.project1)
|
|
assert not result.has_project_membership(self.project2)
|
|
|
|
# member_user should not have visibility to other teams or projects
|
|
assert not result.has_global_access
|
|
assert not result.has_team_access(self.team2)
|
|
assert not result.has_project_access(self.project2)
|
|
|
|
def test_member_role_in_organization_open_membership(self):
|
|
with assume_test_silo_mode(SiloMode.REGION):
|
|
self.org.flags.allow_joinleave = True
|
|
self.org.save()
|
|
member_user = self.create_user(is_superuser=False)
|
|
self.create_member(
|
|
user=member_user, organization=self.org, role="member", teams=[self.team1]
|
|
)
|
|
|
|
request = self.make_request(member_user, is_superuser=False)
|
|
result = self.from_request(request, self.org)
|
|
|
|
assert result.role == "member"
|
|
assert result.team_ids_with_membership == frozenset({self.team1.id})
|
|
assert result.has_team_access(self.team1)
|
|
assert result.project_ids_with_team_membership == frozenset({self.project1.id})
|
|
assert result.has_project_access(self.project1)
|
|
assert result.has_project_membership(self.project1)
|
|
assert not result.has_project_membership(self.project2)
|
|
|
|
# member_user should have visibility to other teams or projects
|
|
assert result.has_global_access
|
|
assert result.has_team_access(self.team2)
|
|
assert result.has_project_access(self.project2)
|
|
|
|
def test_with_valid_auth(self):
|
|
user = self.create_user()
|
|
organization = self.create_organization()
|
|
|
|
member_team = self.create_team(organization=organization)
|
|
member_project = self.create_project(organization=organization, teams=[member_team])
|
|
non_member_team = self.create_team(organization=organization)
|
|
non_member_project = self.create_project(organization=organization, teams=[non_member_team])
|
|
self.create_member(user=user, organization=organization, role="admin", teams=[member_team])
|
|
|
|
request = self.make_request()
|
|
request.auth = self.create_api_key(organization=organization, allowed_origins="*")
|
|
result = self.from_request(request, organization)
|
|
|
|
assert result.team_ids_with_membership == frozenset({})
|
|
assert result.has_team_access(member_team)
|
|
assert result.has_team_access(non_member_team)
|
|
assert result.project_ids_with_team_membership == frozenset({})
|
|
assert result.has_project_access(member_project)
|
|
assert result.has_project_access(non_member_project)
|
|
assert result.has_project_membership(member_project) is False
|
|
assert result.has_project_membership(non_member_project) is False
|
|
assert result.has_global_access
|
|
|
|
def test_with_invalid_auth(self):
|
|
self.create_user()
|
|
organization = self.create_organization()
|
|
other_organization = self.create_organization()
|
|
|
|
team = self.create_team(organization=organization)
|
|
project = self.create_project(organization=organization, teams=[team])
|
|
|
|
request = self.make_request()
|
|
# Using an API key for another org should be invalid
|
|
request.auth = self.create_api_key(organization=other_organization, allowed_origins="*")
|
|
result = self.from_request(request, organization)
|
|
|
|
assert result == NoAccess()
|
|
|
|
assert result.team_ids_with_membership == frozenset({})
|
|
assert result.has_team_access(team) is False
|
|
assert result.project_ids_with_team_membership == frozenset({})
|
|
assert result.has_project_access(project) is False
|
|
assert result.has_project_membership(project) is False
|
|
assert result.has_global_access is False
|
|
|
|
|
|
@all_silo_test
|
|
class FromSentryAppTest(AccessFactoryTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
|
|
# Partner's normal Sentry account.
|
|
self.user = self.create_user("[email protected]")
|
|
|
|
self.org = self.create_organization()
|
|
self.org2 = self.create_organization()
|
|
self.out_of_scope_org = self.create_organization()
|
|
|
|
self.team = self.create_team(organization=self.org)
|
|
self.team2 = self.create_team(organization=self.org2)
|
|
self.out_of_scope_team = self.create_team(organization=self.out_of_scope_org)
|
|
|
|
self.project = self.create_project(organization=self.org, teams=[self.team])
|
|
self.out_of_scope_project = self.create_project(
|
|
organization=self.out_of_scope_org, teams=[self.out_of_scope_team]
|
|
)
|
|
|
|
self.sentry_app = self.create_sentry_app(name="SlowDB", organization=self.org)
|
|
self.out_of_scope_sentry_app = self.create_sentry_app(
|
|
name="Other App", organization=self.out_of_scope_org
|
|
)
|
|
|
|
self.proxy_user = self.sentry_app.proxy_user
|
|
self.out_of_scope_proxy_user = self.out_of_scope_sentry_app.proxy_user
|
|
|
|
self.install = self.create_sentry_app_installation(
|
|
organization=self.org, slug=self.sentry_app.slug, user=self.user
|
|
)
|
|
self.install2 = self.create_sentry_app_installation(
|
|
organization=self.org2, slug=self.sentry_app.slug, user=self.user
|
|
)
|
|
|
|
def test_has_access(self):
|
|
request = self.make_request(user=self.proxy_user)
|
|
result = self.from_request(request, self.org)
|
|
assert result.has_global_access
|
|
assert result.has_team_access(self.team)
|
|
assert result.team_ids_with_membership == frozenset({self.team.id})
|
|
assert result.scopes == frozenset()
|
|
assert result.has_project_access(self.project)
|
|
assert result.has_project_membership(self.project)
|
|
assert not result.has_project_access(self.out_of_scope_project)
|
|
assert not result.permissions
|
|
|
|
def test_no_access_due_to_no_app(self):
|
|
user = self.create_user("[email protected]")
|
|
request = self.make_request(user=user)
|
|
result = self.from_request(request, self.org)
|
|
assert not result.has_team_access(self.team)
|
|
assert not result.has_team_access(self.team2)
|
|
assert not result.has_team_access(self.out_of_scope_team)
|
|
assert not result.has_project_access(self.project)
|
|
assert not result.has_project_access(self.out_of_scope_project)
|
|
|
|
def test_no_access_due_to_no_installation_unowned(self):
|
|
request = self.make_request(user=self.proxy_user)
|
|
result = self.from_request(request, self.out_of_scope_org)
|
|
assert not result.has_team_access(self.team)
|
|
assert not result.has_team_access(self.team2)
|
|
assert not result.has_team_access(self.out_of_scope_team)
|
|
assert not result.has_project_access(self.project)
|
|
assert not result.has_project_access(self.out_of_scope_project)
|
|
|
|
def test_no_access_due_to_no_installation_owned(self):
|
|
request = self.make_request(user=self.out_of_scope_proxy_user)
|
|
result = self.from_request(request, self.out_of_scope_org)
|
|
assert not result.has_team_access(self.team)
|
|
assert not result.has_team_access(self.team2)
|
|
assert not result.has_team_access(self.out_of_scope_team)
|
|
assert not result.has_project_access(self.project)
|
|
assert not result.has_project_access(self.out_of_scope_project)
|
|
|
|
def test_no_access_due_to_invalid_user(self):
|
|
request = self.make_request(user=self.out_of_scope_proxy_user)
|
|
result = self.from_request(request, self.org)
|
|
assert not result.has_team_access(self.team)
|
|
assert not result.has_team_access(self.team2)
|
|
assert not result.has_team_access(self.out_of_scope_team)
|
|
assert not result.has_project_access(self.project)
|
|
assert not result.has_project_access(self.out_of_scope_project)
|
|
|
|
def test_no_deleted_projects(self):
|
|
self.create_member(organization=self.org, user=self.user, role="owner", teams=[self.team])
|
|
deleted_project = self.create_project(
|
|
organization=self.org, status=ObjectStatus.PENDING_DELETION, teams=[self.team]
|
|
)
|
|
request = self.make_request(user=self.proxy_user)
|
|
result = self.from_request(request, self.org)
|
|
assert result.has_project_access(deleted_project) is False
|
|
assert result.has_project_membership(deleted_project) is False
|
|
|
|
def test_no_deleted_teams(self):
|
|
deleted_team = self.create_team(organization=self.org, status=TeamStatus.PENDING_DELETION)
|
|
self.create_member(
|
|
organization=self.org, user=self.user, role="owner", teams=[self.team, deleted_team]
|
|
)
|
|
request = self.make_request(user=self.proxy_user)
|
|
result = self.from_request(request, self.org)
|
|
assert result.has_team_access(deleted_team) is False
|
|
|
|
def test_has_app_scopes(self):
|
|
app_with_scopes = self.create_sentry_app(name="ScopeyTheApp", organization=self.org)
|
|
with assume_test_silo_mode(SiloMode.CONTROL):
|
|
app_with_scopes.update(scope_list=["team:read", "team:write"])
|
|
self.create_sentry_app_installation(
|
|
organization=self.org, slug=app_with_scopes.slug, user=self.user
|
|
)
|
|
|
|
request = self.make_request(user=app_with_scopes.proxy_user)
|
|
result = self.from_request(request, self.org)
|
|
assert result.scopes == frozenset({"team:read", "team:write"})
|
|
assert result.has_scope("team:read") is True
|
|
assert result.has_scope("team:write") is True
|
|
assert result.has_scope("team:admin") is False
|
|
|
|
|
|
@no_silo_test
|
|
class DefaultAccessTest(TestCase):
|
|
def test_no_access(self):
|
|
result = access.DEFAULT
|
|
assert result.sso_is_valid
|
|
assert not result.scopes
|
|
assert not result.has_team_access(Mock())
|
|
assert not result.has_team_scope(Mock(), "project:read")
|
|
assert not result.has_project_access(Mock())
|
|
assert not result.has_projects_access([Mock()])
|
|
assert not result.has_project_scope(Mock(), "project:read")
|
|
assert not result.has_project_membership(Mock())
|
|
assert not result.permissions
|
|
|
|
|
|
@no_silo_test
|
|
class SystemAccessTest(TestCase):
|
|
def test_system_access(self):
|
|
org = self.create_organization()
|
|
team = self.create_team(organization=org)
|
|
project = self.create_project(teams=[team])
|
|
result = access.SystemAccess()
|
|
assert not result.sso_is_valid
|
|
assert not result.requires_sso
|
|
assert result.has_project_access(project)
|
|
assert result.has_any_project_scope(project, "project:read")
|
|
assert not result.has_team_membership(team)
|
|
assert result.has_scope("project:read")
|
|
assert result.has_team_access(team)
|
|
|
|
|
|
@no_silo_test
|
|
class GetPermissionsForUserTest(TestCase):
|
|
def test_combines_roles_and_perms(self):
|
|
user = self.user
|
|
|
|
self.add_user_permission(user, "test.permission")
|
|
role = UserRole.objects.create(name="test.role", permissions=["test.permission-role"])
|
|
role.users.add(user)
|
|
|
|
assert sorted(access_service.get_permissions_for_user(user.id)) == sorted(
|
|
["test.permission", "test.permission-role"]
|
|
)
|