fix(user): filter out soft-deleted memberships and organizations
Add get_active_memberships() method to User model that filters out soft-deleted memberships and memberships of deleted organizations. Update all usages of organization_memberships to use this method, ensuring consistent handling of soft-deleted records across the codebase. Also add deleted_at filters to CA queries in SSH helpers.
This commit is contained in:
@@ -14,13 +14,17 @@ _logger = logging.getLogger(__name__)
|
||||
def _get_org_ca_for_user(user, ca_type: str = "user"):
|
||||
try:
|
||||
from gatehouse_app.models.ssh_ca.ca import CA, CaType
|
||||
org_ids = [m.organization_id for m in user.organization_memberships]
|
||||
|
||||
org_ids = [m.organization_id for m in user.get_active_memberships()]
|
||||
|
||||
if not org_ids:
|
||||
return None
|
||||
|
||||
return CA.query.filter(
|
||||
CA.organization_id.in_(org_ids),
|
||||
CA.ca_type == CaType(ca_type),
|
||||
CA.is_active == True, # noqa: E712
|
||||
CA.is_active == True,
|
||||
CA.deleted_at.is_(None),
|
||||
).first()
|
||||
except Exception:
|
||||
return None
|
||||
@@ -34,7 +38,7 @@ def _get_or_create_system_ca():
|
||||
import os
|
||||
|
||||
try:
|
||||
existing = CA.query.filter_by(name="system-config-ca").first()
|
||||
existing = CA.query.filter_by(name="system-config-ca", deleted_at=None).first()
|
||||
if existing:
|
||||
return existing
|
||||
|
||||
@@ -60,7 +64,7 @@ def _get_or_create_system_ca():
|
||||
|
||||
fingerprint = compute_ssh_fingerprint(pub_key)
|
||||
|
||||
existing_by_fp = CA.query.filter_by(fingerprint=fingerprint).first()
|
||||
existing_by_fp = CA.query.filter_by(fingerprint=fingerprint, deleted_at=None).first()
|
||||
if existing_by_fp:
|
||||
return existing_by_fp
|
||||
|
||||
|
||||
@@ -329,7 +329,7 @@ def admin_hard_delete_user(user_id):
|
||||
if target.id == caller.id:
|
||||
return api_response(success=False, message="Cannot delete your own account via this endpoint.", status=400, error_type="BAD_REQUEST")
|
||||
|
||||
target_org_ids = {m.organization_id for m in target.organization_memberships}
|
||||
target_org_ids = {m.organization_id for m in target.get_active_memberships()}
|
||||
admin_in_shared_org = OrganizationMember.query.filter(
|
||||
OrganizationMember.user_id == caller.id,
|
||||
OrganizationMember.organization_id.in_(target_org_ids),
|
||||
|
||||
@@ -116,9 +116,24 @@ class User(BaseModel):
|
||||
is not None
|
||||
)
|
||||
|
||||
def get_active_memberships(self):
|
||||
"""Get active (non-deleted) organization memberships with active organizations.
|
||||
|
||||
Returns:
|
||||
List of OrganizationMember instances where:
|
||||
- membership.deleted_at is None
|
||||
- organization exists and organization.deleted_at is None
|
||||
"""
|
||||
return [
|
||||
m for m in self.organization_memberships
|
||||
if m.deleted_at is None
|
||||
and m.organization
|
||||
and m.organization.deleted_at is None
|
||||
]
|
||||
|
||||
def get_organizations(self):
|
||||
"""Get all organizations the user is a member of."""
|
||||
return [membership.organization for membership in self.organization_memberships]
|
||||
"""Get all active organizations the user is a member of."""
|
||||
return [membership.organization for membership in self.get_active_memberships()]
|
||||
|
||||
def has_totp_enabled(self) -> bool:
|
||||
"""Check if user has TOTP enabled and verified.
|
||||
|
||||
@@ -55,9 +55,9 @@ def get_userinfo(access_token: str, validate_access_token_fn) -> Dict:
|
||||
|
||||
def _get_user_roles(user: User) -> list:
|
||||
roles = []
|
||||
if not user or not user.organization_memberships:
|
||||
if not user:
|
||||
return roles
|
||||
for member in user.organization_memberships:
|
||||
for member in user.get_active_memberships():
|
||||
roles.append({
|
||||
"organization_id": str(member.organization_id),
|
||||
"role": member.role.value,
|
||||
|
||||
@@ -324,8 +324,8 @@ class OIDCTokenService:
|
||||
List of role objects with organization_id and role
|
||||
"""
|
||||
roles = []
|
||||
if user and user.organization_memberships:
|
||||
for member in user.organization_memberships:
|
||||
if user:
|
||||
for member in user.get_active_memberships():
|
||||
roles.append({
|
||||
"organization_id": str(member.organization_id),
|
||||
"role": member.role.value
|
||||
|
||||
Reference in New Issue
Block a user