fix:handled the super adminlogin
This commit is contained in:
@@ -42,13 +42,15 @@ def _validate_saas_subscription(subscription_details: Optional[dict]) -> None:
|
|||||||
except ValueError:
|
except ValueError:
|
||||||
end_date = None
|
end_date = None
|
||||||
|
|
||||||
if is_active is False or status_value in {"INACTIVE", "EXPIRED"}:
|
can_sign_in = subscription_details.get("can_sign_in", True)
|
||||||
|
if can_sign_in is False or is_active is False or status_value in {"INACTIVE", "EXPIRED"}:
|
||||||
raise HTTPException(status_code=403, detail="Tenant subscription is inactive")
|
raise HTTPException(status_code=403, detail="Tenant subscription is inactive")
|
||||||
|
|
||||||
if start_date and today < start_date:
|
if not (can_sign_in and is_active and status_value == "ACTIVE"):
|
||||||
raise HTTPException(status_code=403, detail="Tenant subscription is not active yet")
|
if start_date and today < start_date:
|
||||||
|
raise HTTPException(status_code=403, detail="Tenant subscription is not active yet")
|
||||||
|
|
||||||
if end_date and today > end_date:
|
if end_date and today > end_date and not subscription_details.get("can_write", True):
|
||||||
raise HTTPException(status_code=403, detail="Tenant subscription has expired")
|
raise HTTPException(status_code=403, detail="Tenant subscription has expired")
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -54,19 +54,22 @@ def _assert_role_in_callers_tenant(role, current_user: User, db: Session = None,
|
|||||||
if role.tenant_id == current_user.tenant_id:
|
if role.tenant_id == current_user.tenant_id:
|
||||||
return
|
return
|
||||||
|
|
||||||
if allow_system_roles and getattr(role, "is_system", False) and current_user.tenant_id and db:
|
if allow_system_roles:
|
||||||
from app.modules.billing.models.plan_model import TenantSubscription, PlanRole
|
if getattr(role, "tenant_id", None) is None and getattr(role, "name", "").lower() != "superadmin":
|
||||||
sub = db.query(TenantSubscription).filter(
|
return
|
||||||
TenantSubscription.tenant_id == current_user.tenant_id,
|
if getattr(role, "is_system", False) and current_user.tenant_id and db:
|
||||||
TenantSubscription.status == 'active'
|
from app.modules.billing.models.plan_model import TenantSubscription, PlanRole
|
||||||
).order_by(TenantSubscription.created_at.desc()).first()
|
sub = db.query(TenantSubscription).filter(
|
||||||
if sub and sub.plan_id:
|
TenantSubscription.tenant_id == current_user.tenant_id,
|
||||||
has_role = db.query(PlanRole).filter(
|
TenantSubscription.status == 'active'
|
||||||
PlanRole.plan_id == sub.plan_id,
|
).order_by(TenantSubscription.created_at.desc()).first()
|
||||||
PlanRole.role_id == role.id
|
if sub and sub.plan_id:
|
||||||
).first()
|
has_role = db.query(PlanRole).filter(
|
||||||
if has_role:
|
PlanRole.plan_id == sub.plan_id,
|
||||||
return
|
PlanRole.role_id == role.id
|
||||||
|
).first()
|
||||||
|
if has_role:
|
||||||
|
return
|
||||||
|
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_404_NOT_FOUND, detail="Role not found"
|
status_code=status.HTTP_404_NOT_FOUND, detail="Role not found"
|
||||||
|
|||||||
@@ -63,6 +63,7 @@ class RoleService:
|
|||||||
query = db.query(Role)
|
query = db.query(Role)
|
||||||
if tenant_id is not None:
|
if tenant_id is not None:
|
||||||
from app.modules.billing.models.plan_model import TenantSubscription, PlanRole
|
from app.modules.billing.models.plan_model import TenantSubscription, PlanRole
|
||||||
|
from sqlalchemy import and_, func
|
||||||
sub = db.query(TenantSubscription).filter(
|
sub = db.query(TenantSubscription).filter(
|
||||||
TenantSubscription.tenant_id == tenant_id,
|
TenantSubscription.tenant_id == tenant_id,
|
||||||
TenantSubscription.status == 'active'
|
TenantSubscription.status == 'active'
|
||||||
@@ -72,7 +73,16 @@ class RoleService:
|
|||||||
plan_role_ids = [pr.role_id for pr in plan_roles]
|
plan_role_ids = [pr.role_id for pr in plan_roles]
|
||||||
query = query.filter(or_(Role.tenant_id == tenant_id, Role.id.in_(plan_role_ids)))
|
query = query.filter(or_(Role.tenant_id == tenant_id, Role.id.in_(plan_role_ids)))
|
||||||
else:
|
else:
|
||||||
query = query.filter(Role.tenant_id == tenant_id)
|
# Include tenant roles AND tenant-accessible default/system roles (excluding superadmin)
|
||||||
|
query = query.filter(
|
||||||
|
or_(
|
||||||
|
Role.tenant_id == tenant_id,
|
||||||
|
and_(
|
||||||
|
Role.tenant_id.is_(None),
|
||||||
|
func.lower(Role.name) != 'superadmin'
|
||||||
|
)
|
||||||
|
)
|
||||||
|
)
|
||||||
return query.all()
|
return query.all()
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
@@ -135,6 +145,7 @@ class RoleService:
|
|||||||
query = db.query(Role)
|
query = db.query(Role)
|
||||||
if tenant_id is not None:
|
if tenant_id is not None:
|
||||||
from app.modules.billing.models.plan_model import TenantSubscription, PlanRole
|
from app.modules.billing.models.plan_model import TenantSubscription, PlanRole
|
||||||
|
from sqlalchemy import and_, func
|
||||||
sub = db.query(TenantSubscription).filter(
|
sub = db.query(TenantSubscription).filter(
|
||||||
TenantSubscription.tenant_id == tenant_id,
|
TenantSubscription.tenant_id == tenant_id,
|
||||||
TenantSubscription.status == 'active'
|
TenantSubscription.status == 'active'
|
||||||
@@ -144,7 +155,16 @@ class RoleService:
|
|||||||
plan_role_ids = [pr.role_id for pr in plan_roles]
|
plan_role_ids = [pr.role_id for pr in plan_roles]
|
||||||
query = query.filter(or_(Role.tenant_id == tenant_id, Role.id.in_(plan_role_ids)))
|
query = query.filter(or_(Role.tenant_id == tenant_id, Role.id.in_(plan_role_ids)))
|
||||||
else:
|
else:
|
||||||
query = query.filter(Role.tenant_id == tenant_id)
|
# Include tenant roles AND tenant-accessible default/system roles (excluding superadmin)
|
||||||
|
query = query.filter(
|
||||||
|
or_(
|
||||||
|
Role.tenant_id == tenant_id,
|
||||||
|
and_(
|
||||||
|
Role.tenant_id.is_(None),
|
||||||
|
func.lower(Role.name) != 'superadmin'
|
||||||
|
)
|
||||||
|
)
|
||||||
|
)
|
||||||
if search and search.strip():
|
if search and search.strip():
|
||||||
search_term = search.strip()
|
search_term = search.strip()
|
||||||
query = query.filter(Role.name.ilike(f"%{search_term}%"))
|
query = query.filter(Role.name.ilike(f"%{search_term}%"))
|
||||||
|
|||||||
Reference in New Issue
Block a user