mirror of
https://github.com/hanzoai/insights.git
synced 2026-10-30 16:15:31 +00:00
zeekay
b7f9a67e2a
The rebase pulled upstream files that import ee/, a tree this distribution does not carry, so `django.setup()` raised ModuleNotFoundError: No module named 'ee' before any request could be served. main had already severed this; the rebase lost the severance. Restore insights/models/ee_models.py: the four RBAC models upstream keeps under ee/, mapped managed=False onto the tables that actually exist (ee_accesscontrol, ee_dashboardprivilege, ee_role, ee_rolemembership -- verified against the live database, which has those four and no other ee_ table). 131 files repointed at it. Restore main's insights/rbac/access_control_api_mixin.py. The rebase left a fallback of `class AccessControlViewSetMixin: pass`, which drops the dangerously_get_required_scopes hook that insights/api/project.py, insights/api/team.py and products/replay_vision reach through super() -- an AttributeError on scope derivation. Enforcement itself is unaffected either way: it lives in insights/rbac/user_access_control.py, in-tree. ExplicitTeamMembership has no table in this distribution, so it cannot emit the signal the org serializer cache subscribes to; drop the registration. SyncIterableToAsync moves in-tree next to the other streaming helpers, written from the interface rather than carried over, so the MCP store proxy no longer reaches into ee/ for a generic ASGI adapter. bin/debrand renamed the hog_flow directory and the HogFlow symbol but not the module path between them, leaving `from .hog_flow.hog_flow import InsightsFlow` pointing at a path that no longer exists (33 sites). Add the missing text rule so the rename is closed, and apply it.
63 lines
2.4 KiB
Python
63 lines
2.4 KiB
Python
from typing import cast
|
|
|
|
import structlog
|
|
|
|
from insights.models import Team, User
|
|
from insights.rbac.user_access_control import UserAccessControl
|
|
from insights.scopes import APIScopeObject
|
|
|
|
from products.notifications.backend.facade.enums import TargetType
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
|
|
class RecipientsResolver:
|
|
def resolve(self, target_type: TargetType, target_id: str, team_id: int | None) -> list[int]:
|
|
if target_type == TargetType.USER:
|
|
return [int(target_id)]
|
|
elif target_type == TargetType.TEAM:
|
|
team = Team.objects.filter(id=int(target_id)).first()
|
|
if team is None:
|
|
return []
|
|
return list(team.all_users_with_access().values_list("id", flat=True))
|
|
elif target_type == TargetType.ORGANIZATION:
|
|
from insights.models import OrganizationMembership
|
|
|
|
return list(
|
|
OrganizationMembership.objects.filter(
|
|
organization_id=target_id,
|
|
).values_list("user_id", flat=True)
|
|
)
|
|
elif target_type == TargetType.ROLE:
|
|
from insights.models.ee_models import RoleMembership
|
|
|
|
return list(
|
|
RoleMembership.objects.filter(
|
|
role_id=target_id,
|
|
).values_list("user_id", flat=True)
|
|
)
|
|
|
|
raise ValueError(f"Unknown target type: {target_type}")
|
|
|
|
def filter_by_access_control(self, user_ids: list[int], resource_type: str, team: Team) -> list[int]:
|
|
"""Filter user IDs by access control. Not overridable — always applied after resolve()."""
|
|
try:
|
|
sample_user = User.objects.filter(id__in=user_ids).first()
|
|
if not sample_user:
|
|
return user_ids
|
|
sample_ac = UserAccessControl(sample_user, team)
|
|
if not sample_ac.access_controls_supported:
|
|
return user_ids
|
|
except Exception:
|
|
logger.exception("notifications.ac_check_failed")
|
|
return user_ids
|
|
|
|
filtered = []
|
|
# TODO: optimize for large orgs — batch AC checks instead of per-user
|
|
users = User.objects.filter(id__in=user_ids)
|
|
for user in users:
|
|
ac = UserAccessControl(user, team)
|
|
if ac.check_access_level_for_resource(cast(APIScopeObject, resource_type), "viewer"):
|
|
filtered.append(user.id)
|
|
return filtered
|