diff --git a/apps/employees/admin.py b/apps/employees/admin.py index 30d7297..900da2b 100644 --- a/apps/employees/admin.py +++ b/apps/employees/admin.py @@ -541,10 +541,21 @@ if admin.site.is_registered(Group): admin.site.unregister(Group) +from apps.locations.models import GroupLocationAccess + + +class GroupLocationAccessInline(TabularInline): + model = GroupLocationAccess + extra = 1 + max_num = 1 + filter_horizontal = ('locations',) + + @admin.register(Group) class GroupAdmin(ModelAdmin): """Роль = группа доступа: имя, портальные разделы, прочие права, описание/активна.""" form = GroupAdminForm + inlines = [GroupLocationAccessInline] filter_horizontal = ('permissions',) search_fields = ('name',) list_display = ('name', 'active_display', 'description_display', 'portal_matrix_link') diff --git a/apps/it_manager/billing.py b/apps/it_manager/billing.py index e39707a..2316fc8 100644 --- a/apps/it_manager/billing.py +++ b/apps/it_manager/billing.py @@ -12,6 +12,7 @@ from django.views.decorators.http import require_POST from apps.it_manager.models import InternetContract, InternetContractPayment, InternetProvider from core.portal_access import section_required +from core.location_access import scoped_queryset UKR_MONTHS = ['', 'Січень', 'Лютий', 'Березень', 'Квітень', 'Травень', 'Червень', 'Липень', 'Серпень', 'Вересень', 'Жовтень', 'Листопад', 'Грудень'] @@ -27,15 +28,15 @@ def _parse_period(request) -> datetime.date: return today.replace(day=1) -def _contracts_queryset(provider_id=None): +def _contracts_queryset(provider_id=None, *, user): qs = ( - InternetContract.objects.select_related( + scoped_queryset(user, InternetContract.objects.all()).select_related( 'provider', 'location', 'parent_contract', 'status', ) .prefetch_related( Prefetch( 'child_contracts', - queryset=InternetContract.objects.select_related('location', 'status').order_by('contract_number'), + queryset=scoped_queryset(user, InternetContract.objects.all()).select_related('location', 'status').order_by('contract_number'), ), ) .filter(status__code='active') @@ -154,7 +155,7 @@ def _internet_billing_context(request, **extra): next_period = datetime.date(period.year, period.month + 1, 1) provider_id = request.GET.get('provider') or request.POST.get('provider') or '' - contracts = list(_contracts_queryset(provider_id or None)) + contracts = list(_contracts_queryset(provider_id or None, user=request.user)) contract_ids = [c.pk for c in contracts] payments = InternetContractPayment.objects.filter( @@ -178,7 +179,7 @@ def _internet_billing_context(request, **extra): providers = list( InternetProvider.objects.filter(is_active=True) - .filter(contracts__status__code='active') + .filter(contracts__in=_contracts_queryset(user=request.user)) .distinct() .order_by('name') ) @@ -224,7 +225,7 @@ def internet_payment_update_view(request): if not contract_id: return render_accounting_portal(request, 'internet', billing_form_error='Не указан договор.') - contract = InternetContract.objects.filter(pk=contract_id).first() + contract = scoped_queryset(request.user, InternetContract.objects.all()).filter(pk=contract_id).first() if contract is None: return render_accounting_portal(request, 'internet', billing_form_error='Договор не найден.') diff --git a/apps/it_manager/tests.py b/apps/it_manager/tests.py index b5ab5b3..c991d1a 100644 --- a/apps/it_manager/tests.py +++ b/apps/it_manager/tests.py @@ -10,7 +10,7 @@ from apps.it_manager.models import ( Equipment, EquipmentAssignment, EquipmentModel, Ticket, WorkLog, ) from core.models.directories import Directory -from core.test_utils import login_2fa +from core.test_utils import login_2fa, grant_all_locations Employee = get_user_model() @@ -89,6 +89,7 @@ class ItManagerEquipmentTests(TestCase): def setUp(self): self.it_user = make_employee(is_staff=False) self.it_user.groups.add(Group.objects.get(name='IT')) + grant_all_locations(self.it_user) self.client = Client() login_2fa(self.client, self.it_user) @@ -274,6 +275,7 @@ class ItManagerTicketTests(TestCase): def setUp(self): self.it_user = make_employee(is_staff=False) self.it_user.groups.add(Group.objects.get(name='IT')) + grant_all_locations(self.it_user) self.requester = make_employee() self.client = Client() login_2fa(self.client, self.it_user) @@ -311,6 +313,7 @@ class ItManagerWorkLogTests(TestCase): def setUp(self): self.it_user = make_employee(is_staff=False) self.it_user.groups.add(Group.objects.get(name='IT')) + grant_all_locations(self.it_user) self.client = Client() login_2fa(self.client, self.it_user) @@ -398,6 +401,7 @@ class ItManagerWorkLogFilterEditTests(TestCase): it_group = Group.objects.get(name='IT') for u in (self.head, self.tech_a, self.tech_b): u.groups.add(it_group) + grant_all_locations(u) self.log_a = WorkLog.objects.create( date=date(2026, 9, 1), diff --git a/apps/it_manager/tests_billing.py b/apps/it_manager/tests_billing.py index 259b603..50bf95f 100644 --- a/apps/it_manager/tests_billing.py +++ b/apps/it_manager/tests_billing.py @@ -10,7 +10,7 @@ from apps.it_manager.models import InternetContract, InternetContractPayment, In from apps.locations.models.base import Location from core.models.directories import Directory, DirectoryType from core.portal_access import user_can_access -from core.test_utils import login_2fa +from core.test_utils import login_2fa, grant_all_locations from core.tests_portal_access import make_employee @@ -170,7 +170,7 @@ class InternetBillingPortalTests(TestCase): start_date=date(2024, 1, 1), monthly_cost=Decimal('500.00'), ) - self.user = make_employee() + self.user = grant_all_locations(make_employee()) self.user.groups.add(Group.objects.get(name='Офисный сотрудник')) for attr in ('_perm_cache', '_user_perm_cache', '_group_perm_cache'): if hasattr(self.user, attr): diff --git a/apps/it_manager/views.py b/apps/it_manager/views.py index fae34bb..dbe4b8e 100644 --- a/apps/it_manager/views.py +++ b/apps/it_manager/views.py @@ -28,13 +28,52 @@ from core.portal_access import deny_section_access from core.utils.htmx import render_page from core.utils.pagination import paginate +from core.location_access import ( + accessible_locations, scoped_queryset, scoped_employees, has_all_locations, require_location, +) +from functools import wraps +from django.http import Http404 + Employee = get_user_model() -def _it_locations_qs(): +def _location_scoped_post(view): + """Проверить все присланные ссылки до любых изменений, включая bulk.""" + @wraps(view) + def wrapped(request, *args, **kwargs): + if request.method == 'POST' and not has_all_locations(request.user): + for field, model in [('location_id', Location), ('equipment_id', Equipment), + ('ticket_id', Ticket), ('parent_contract_id', InternetContract)]: + value = request.POST.get(field) + if value: + if not value.isdigit(): + raise Http404 + qs = accessible_locations(request.user) if model is Location else scoped_queryset(request.user, model.objects.all()) + get_object_or_404(qs, pk=value) + for field in ('assigned_to_id', 'caller_id'): + value = request.POST.get(field) + if value: + if not value.isdigit(): + raise Http404 + get_object_or_404(scoped_employees(request.user, Employee.objects.all()), pk=value) + ids = request.POST.getlist('equipment_ids') + if ids: + if any(not value.isdigit() for value in ids): + raise Http404 + if scoped_queryset(request.user, Equipment.objects.all()).filter(pk__in=ids).count() != len(set(ids)): + raise Http404 + # Общий склад без локации и рамочные договоры относятся ко всей сети. + if request.POST.get('is_framework'): + return HttpResponseForbidden('Рамочные договоры требуют доступа ко всей сети.') + return view(request, *args, **kwargs) + return wrapped + + + +def _it_locations_qs(user): """Локации для селектов IT: аптеки с организацией (чтобы отличать №12 у разных юрлиц).""" return ( - Location.objects.filter(is_active=True) + accessible_locations(user).filter(is_active=True) .select_related('location_region') .prefetch_related( models.Prefetch( @@ -75,6 +114,8 @@ def _parse_assign_post(request) -> tuple[str, str | None, str | None, str | None assigned_to_id = request.POST.get('assigned_to_id') or None if not target: + if not has_all_locations(request.user): + return '', None, None, 'Выберите доступную локацию или сотрудника.' return '', None, None, None if target == 'employee': @@ -87,7 +128,7 @@ def _parse_assign_post(request) -> tuple[str, str | None, str | None, str | None return target, None, None, ( 'Оберіть аптеку.' if target == 'pharmacy' else 'Оберіть офіс / відділ.' ) - loc = Location.objects.filter(pk=location_id, is_active=True).first() + loc = accessible_locations(request.user).filter(pk=location_id, is_active=True).first() if not loc: return target, None, None, 'Локацію не знайдено.' if target == 'pharmacy' and loc.location_type != 'pharmacy': @@ -182,7 +223,7 @@ def _apply_equipment_issue(unit, *, location_id=None, assigned_to_id=None, user= def _it_manager_context(request): equipment_qs = ( - Equipment.objects.select_related( + scoped_queryset(request.user, Equipment.objects.all()).select_related( 'equipment_model__brand', 'equipment_model__equipment_type', 'assigned_to', 'location', 'status', ) @@ -197,14 +238,14 @@ def _it_manager_context(request): equipment_models = list( EquipmentModel.objects.select_related('equipment_type', 'brand') - .annotate(units_total=models.Count('units')) + .annotate(units_total=models.Count('units', filter=Q(units__in=scoped_queryset(request.user, Equipment.objects.all())))) .order_by('equipment_type__name', 'brand__name', 'name') ) - locations_all = list(_it_locations_qs()) + locations_all = list(_it_locations_qs(request.user)) it_pharmacies, it_offices = _split_it_locations(locations_all) tickets_qs = ( - Ticket.objects.select_related( + scoped_queryset(request.user, Ticket.objects.all()).select_related( 'requester', 'assignee', 'equipment', 'location', 'ticket_type', 'priority', 'status', ) @@ -212,7 +253,7 @@ def _it_manager_context(request): 'comments__author', models.Prefetch( 'work_logs', - queryset=WorkLog.objects.select_related('technician', 'work_type').order_by('-date', '-created_at'), + queryset=scoped_queryset(request.user, WorkLog.objects.all()).select_related('technician', 'work_type').order_by('-date', '-created_at'), ), ) .annotate(worklogs_count=models.Count('work_logs', distinct=True)) @@ -250,7 +291,7 @@ def _it_manager_context(request): worklog_month_id = 'all' worklogs_qs = ( - WorkLog.objects.select_related( + scoped_queryset(request.user, WorkLog.objects.all()).select_related( 'technician', 'location', 'equipment', 'ticket', 'work_type', ) .prefetch_related( @@ -279,7 +320,7 @@ def _it_manager_context(request): stats_start, stats_end = month_start, None worklog_stats_label = 'этом месяце' - month_logs = WorkLog.objects.filter(date__gte=stats_start) + month_logs = scoped_queryset(request.user, WorkLog.objects.all()).filter(date__gte=stats_start) if stats_end is not None: month_logs = month_logs.filter(date__lte=stats_end) if worklog_technician_id: @@ -289,7 +330,7 @@ def _it_manager_context(request): worklog_month_choices = [] seen_months = set() - for d in WorkLog.objects.dates('date', 'month', order='DESC'): + for d in scoped_queryset(request.user, WorkLog.objects.all()).dates('date', 'month', order='DESC'): mid = f'{d.year:04d}-{d.month:02d}' seen_months.add(mid) worklog_month_choices.append((mid, f'{MONTHS_RU[d.month]} {d.year}')) @@ -303,7 +344,7 @@ def _it_manager_context(request): worklog_technicians = ( Employee.objects.filter(is_active=True) .filter( - Q(pk__in=WorkLog.objects.values('technician_id')) + Q(pk__in=scoped_queryset(request.user, WorkLog.objects.all()).values('technician_id')) | Q(groups__name='IT') | Q(is_staff=True) | Q(is_superuser=True) @@ -314,7 +355,7 @@ def _it_manager_context(request): can_edit_worklogs = user_can_manage_it_monthly_report(request.user) consumables_qs = ( - Consumable.objects.select_related('consumable_type', 'color', 'status', 'equipment', 'location') + scoped_queryset(request.user, Consumable.objects.all()).select_related('consumable_type', 'color', 'status', 'equipment', 'location') .order_by('equipment', 'consumable_type__name', 'color__name') ) consumables = list(consumables_qs) @@ -325,10 +366,10 @@ def _it_manager_context(request): providers = list(InternetProvider.objects.order_by('name')) contracts_qs = ( - InternetContract.objects.select_related( + scoped_queryset(request.user, InternetContract.objects.all()).select_related( 'provider', 'location', 'service_type', 'connection_type', 'status', 'parent_contract', ) - .prefetch_related('child_contracts') + .prefetch_related(models.Prefetch('child_contracts', queryset=scoped_queryset(request.user, InternetContract.objects.all()))) .order_by('provider__name', 'contract_number') ) contracts = list(contracts_qs) @@ -336,7 +377,7 @@ def _it_manager_context(request): contracts_page = paginate(request, contracts_qs, page_param='contracts_page', size_param='contracts_per_page') service_calls_qs = ( - ServiceCall.objects.select_related('caller', 'handler', 'location', 'channel', 'ticket') + scoped_queryset(request.user, ServiceCall.objects.all()).select_related('caller', 'handler', 'location', 'channel', 'ticket') .order_by('-date_time') ) service_calls_page = paginate( @@ -355,7 +396,7 @@ def _it_manager_context(request): 'it_locations': locations_all, 'it_pharmacies': it_pharmacies, 'it_offices': it_offices, - 'it_employees': Employee.objects.filter(is_active=True).order_by('last_name', 'first_name'), + 'it_employees': scoped_employees(request.user, Employee.objects.filter(is_active=True)).order_by('last_name', 'first_name'), 'tickets': tickets, 'tickets_page': tickets_page, @@ -394,7 +435,7 @@ def _it_manager_context(request): 'contract_connection_types': _directory_choices('contract_connection_type'), 'contract_statuses': _directory_choices('contract_status'), 'framework_contracts': list( - InternetContract.objects.filter(is_framework=True) + scoped_queryset(request.user, InternetContract.objects.all()).filter(is_framework=True) .select_related('provider') .order_by('provider__name', 'contract_number') ), @@ -408,6 +449,7 @@ def _it_manager_context(request): @login_required +@_location_scoped_post def it_manager_view(request): if not _is_it_staff(request.user): return deny_section_access(request) @@ -428,6 +470,7 @@ def it_manager_view(request): @login_required +@_location_scoped_post def equipment_model_create_view(request): """Додати модель у каталог (тип + бренд + назва). Далі за нею заводяться екземпляри лише з серійниками.""" @@ -459,6 +502,7 @@ def equipment_model_create_view(request): @login_required +@_location_scoped_post def equipment_bulk_create_view(request): """Масовий ввід екземплярів: обираємо модель з каталогу і вставляємо список серійників (по одному в рядку) — створюється N одиниць на склад. @@ -513,7 +557,7 @@ def equipment_bulk_create_view(request): or _resolve_directory('equipment_status', 'stock') existing = set( - Equipment.objects.filter(serial_number__in=serials) + scoped_queryset(request.user, Equipment.objects.all()).filter(serial_number__in=serials) .values_list('serial_number', flat=True) ) created = 0 @@ -552,6 +596,7 @@ def equipment_bulk_create_view(request): @login_required +@_location_scoped_post def equipment_bulk_assign_view(request): """Масова видача: тип (аптека / офіс / співробітник) → список.""" if not _is_it_staff(request.user): @@ -569,7 +614,7 @@ def equipment_bulk_assign_view(request): elif assign_err: context_extra['equipment_form_error'] = assign_err else: - units = Equipment.objects.filter(pk__in=ids) + units = scoped_queryset(request.user, Equipment.objects.all()).filter(pk__in=ids) count = 0 for unit in units: if _apply_equipment_issue( @@ -590,12 +635,13 @@ def equipment_bulk_assign_view(request): @login_required +@_location_scoped_post def equipment_update_view(request, pk): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': - equipment = get_object_or_404(Equipment, pk=pk) + equipment = get_object_or_404(scoped_queryset(request.user, Equipment.objects.all()), pk=pk) new_status_code = request.POST.get('status') or '' new_status_id = _resolve_directory('equipment_status', new_status_code) @@ -603,6 +649,8 @@ def equipment_update_view(request, pk): # Повернення на склад: статус stock або порожній тип видачі if new_status_code == 'stock' or not target: + if not has_all_locations(request.user): + return HttpResponseForbidden('Общий склад требует доступа ко всей сети.') _close_open_assignments(equipment) if new_status_code == 'stock' and new_status_id: equipment.status_id = new_status_id @@ -639,12 +687,13 @@ def equipment_update_view(request, pk): @login_required +@_location_scoped_post def ticket_update_view(request, pk): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': - ticket = get_object_or_404(Ticket, pk=pk) + ticket = get_object_or_404(scoped_queryset(request.user, Ticket.objects.all()), pk=pk) new_status_id = _resolve_directory('ticket_status', request.POST.get('status')) if new_status_id: @@ -659,12 +708,13 @@ def ticket_update_view(request, pk): @login_required +@_location_scoped_post def ticket_comment_view(request, pk): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': - ticket = get_object_or_404(Ticket, pk=pk) + ticket = get_object_or_404(scoped_queryset(request.user, Ticket.objects.all()), pk=pk) text = request.POST.get('text', '').strip() if text: TicketComment.objects.create( @@ -678,6 +728,7 @@ def ticket_comment_view(request, pk): @login_required +@_location_scoped_post def worklog_create_view(request): if not _is_it_staff(request.user): return HttpResponseForbidden() @@ -704,6 +755,8 @@ def worklog_create_view(request): hours_spent = None location_id = request.POST.get('location_id') or None + if not location_id and not has_all_locations(request.user): + return HttpResponseForbidden('Укажите доступную локацию.') equipment_id = request.POST.get('equipment_id') or None ticket_id = request.POST.get('ticket_id') or None @@ -737,6 +790,7 @@ def worklog_create_view(request): @login_required +@_location_scoped_post def worklog_update_view(request, pk): """Inline-редактирование записи журнала — только руководителю отдела.""" if not _is_it_staff(request.user): @@ -746,7 +800,7 @@ def worklog_update_view(request, pk): if request.method != 'POST': return HttpResponseForbidden() - log = get_object_or_404(WorkLog, pk=pk) + log = get_object_or_404(scoped_queryset(request.user, WorkLog.objects.all()), pk=pk) description = request.POST.get('description', '').strip() work_type_raw = request.POST.get('work_type', '') work_type_id = _resolve_directory('worklog_type', work_type_raw) @@ -782,6 +836,8 @@ def worklog_update_view(request, pk): location_raw = (request.POST.get('location_id') or '').strip() location_id = int(location_raw) if location_raw.isdigit() else None + if not location_id and not has_all_locations(request.user): + return HttpResponseForbidden('Укажите доступную локацию.') log.date = log_date log.technician_id = technician_id log.work_type_id = work_type_id @@ -802,11 +858,14 @@ def worklog_update_view(request, pk): @login_required +@_location_scoped_post def consumable_create_view(request): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': + if not request.POST.get('location_id') and not has_all_locations(request.user): + return HttpResponseForbidden('Укажите доступную локацию.') model_number = request.POST.get('model_number', '').strip() consumable_type_id = _resolve_directory('consumable_type', request.POST.get('consumable_type', '')) color_id = _resolve_directory('consumable_color', request.POST.get('color', '')) @@ -830,12 +889,13 @@ def consumable_create_view(request): @login_required +@_location_scoped_post def consumable_update_view(request, pk): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': - consumable = get_object_or_404(Consumable, pk=pk) + consumable = get_object_or_404(scoped_queryset(request.user, Consumable.objects.all()), pk=pk) new_status_code = request.POST.get('status') new_status_id = _resolve_directory('consumable_status', new_status_code) if new_status_id: @@ -850,6 +910,7 @@ def consumable_update_view(request, pk): @login_required +@_location_scoped_post def provider_create_view(request): if not _is_it_staff(request.user): return HttpResponseForbidden() @@ -875,6 +936,7 @@ def provider_create_view(request): @login_required +@_location_scoped_post def contract_create_view(request): if not _is_it_staff(request.user): return HttpResponseForbidden() @@ -919,12 +981,13 @@ def contract_create_view(request): @login_required +@_location_scoped_post def contract_update_view(request, pk): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': - contract = get_object_or_404(InternetContract, pk=pk) + contract = get_object_or_404(scoped_queryset(request.user, InternetContract.objects.all()), pk=pk) new_status_id = _resolve_directory('contract_status', request.POST.get('status')) if new_status_id: contract.status_id = new_status_id @@ -950,11 +1013,14 @@ def contract_update_view(request, pk): @login_required +@_location_scoped_post def servicecall_create_view(request): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': + if not request.POST.get('location_id') and not has_all_locations(request.user): + return HttpResponseForbidden('Укажите доступную локацию.') caller_id = request.POST.get('caller_id') or None description = request.POST.get('description', '').strip() channel_id = _resolve_directory('servicecall_channel', request.POST.get('channel', '')) diff --git a/apps/locations/forms.py b/apps/locations/forms.py index c8bd544..8b4fd1f 100644 --- a/apps/locations/forms.py +++ b/apps/locations/forms.py @@ -69,7 +69,7 @@ class PortalPharmacyEditForm(forms.ModelForm): 'curator': forms.Select(attrs={'class': 'input'}), } - def __init__(self, *args, **kwargs): + def __init__(self, *args, user=None, **kwargs): super().__init__(*args, **kwargs) from apps.employees.models import Employee @@ -98,3 +98,7 @@ class PortalPharmacyEditForm(forms.ModelForm): self.fields['building'].required = False self.fields['apartment'].required = False self.fields['location'].required = False + + if user is not None and not user.has_perm('locations.manage_location_access'): + for field in ('head_pharmacist', 'responsible_manager', 'curator'): + self.fields.pop(field, None) diff --git a/apps/locations/models/__init__.py b/apps/locations/models/__init__.py index c904504..a042ba3 100644 --- a/apps/locations/models/__init__.py +++ b/apps/locations/models/__init__.py @@ -1,3 +1,4 @@ +from .access import GroupLocationAccess from .base import Location, LocationManager from .pharmacies import Pharmacy, PharmacyManager, PharmacyPhone, PharmacyHistory from .dislocation import DislocationColumnPreset @@ -9,7 +10,7 @@ from .pharmacy_visibility import ( ) __all__ = [ - 'Location', 'LocationManager', + 'GroupLocationAccess', 'Location', 'LocationManager', 'Pharmacy', 'PharmacyManager', 'PharmacyPhone', 'PharmacyHistory', 'DislocationColumnPreset', 'PharmacyCardVisibility', 'pharmacy_card_flags', diff --git a/apps/locations/models/base.py b/apps/locations/models/base.py index 7863e4f..13dcff6 100644 --- a/apps/locations/models/base.py +++ b/apps/locations/models/base.py @@ -128,6 +128,7 @@ class Location(BaseModel): objects = LocationManager() class Meta: + permissions = [('view_all_locations', 'Доступ к данным всех локаций портала'), ('manage_location_access', 'Назначение ответственных за локации')] verbose_name = _('Локация') verbose_name_plural = _('Локации') indexes = [ diff --git a/apps/locations/models/pharmacy_visibility.py b/apps/locations/models/pharmacy_visibility.py index 4166dd6..c506ade 100644 --- a/apps/locations/models/pharmacy_visibility.py +++ b/apps/locations/models/pharmacy_visibility.py @@ -104,6 +104,9 @@ def user_can_view_pharmacy_section(user, pharmacy, level: str) -> bool: """Проверка уровня видимости для user × pharmacy.""" if not user or not getattr(user, 'is_authenticated', False): return False + from core.location_access import accessible_pharmacies + if not accessible_pharmacies(user).filter(pk=pharmacy.pk).exists(): + return False if getattr(user, 'is_superuser', False): return True @@ -116,7 +119,7 @@ def user_can_view_pharmacy_section(user, pharmacy, level: str) -> bool: if level == PharmacyCardVisibility.Level.OWN_OR_STAFF: if getattr(user, 'is_staff', False): return True - return getattr(user, 'user_pharmacy_id', None) == getattr(pharmacy, 'pk', None) + return True # область уже проверена: своя или закреплённая аптека if level == PharmacyCardVisibility.Level.IT: from core.portal_access import user_can_access if user_can_access(user, 'it_manager'): diff --git a/apps/locations/tests.py b/apps/locations/tests.py index 8ca6f5e..ee91614 100644 --- a/apps/locations/tests.py +++ b/apps/locations/tests.py @@ -1,3 +1,4 @@ +from core.test_utils import grant_all_locations from unittest import mock from django.core.exceptions import ValidationError @@ -355,6 +356,8 @@ class PharmacyCardVisibilityTests(TestCase): cls.it_user = Employee.objects.create_user(username='vis.it', **common) it_group, _ = Group.objects.get_or_create(name='IT') cls.it_user.groups.add(it_group) + grant_all_locations(cls.it_user) + grant_all_locations(cls.staff) PharmacyCardVisibility.load() def _set_level(self, section, level): @@ -464,7 +467,7 @@ class PharmacyCardVisibilityTests(TestCase): everyone = PharmacyCardVisibility.Level.EVERYONE hidden = PharmacyCardVisibility.Level.HIDDEN - self.assertTrue(user_can_view_pharmacy_section(self.stranger, self.pharmacy, everyone)) + self.assertFalse(user_can_view_pharmacy_section(self.stranger, self.pharmacy, everyone)) self.assertFalse(user_can_view_pharmacy_section(self.staff, self.pharmacy, hidden)) self.assertTrue(user_can_view_pharmacy_section(self.superuser, self.pharmacy, hidden)) @@ -489,7 +492,7 @@ class PharmacyCardVisibilityTests(TestCase): self._set_level('rro', PharmacyCardVisibility.Level.EVERYONE) self._set_level('equipment', PharmacyCardVisibility.Level.HIDDEN) flags = pharmacy_card_flags(self.stranger, self.pharmacy) - self.assertTrue(flags['can_view_rro']) + self.assertFalse(flags['can_view_rro']) self.assertFalse(flags['can_view_equipment']) def test_detail_view_hides_tab_when_section_staff_only(self): diff --git a/apps/locations/views.py b/apps/locations/views.py index 4ba48f3..80f8930 100644 --- a/apps/locations/views.py +++ b/apps/locations/views.py @@ -8,6 +8,8 @@ from django.views.decorators.http import require_POST from core.utils.htmx import render_page +from core.location_access import accessible_pharmacies + from .forms import PortalPharmacyEditForm from .models import Pharmacy, pharmacy_card_flags, user_can_edit_pharmacy @@ -68,7 +70,7 @@ def _pharmacy_detail_context(request, pharmacy, *, form=None, edit_mode=False, can_edit = flags['can_edit'] if edit_mode and can_edit and form is None: - form = PortalPharmacyEditForm(instance=pharmacy) + form = PortalPharmacyEditForm(instance=pharmacy, user=request.user) return { 'pharmacy': pharmacy, @@ -94,7 +96,7 @@ def _pharmacy_detail_context(request, pharmacy, *, form=None, edit_mode=False, @require_POST def pharmacy_update_view(request, pk): pharmacy = get_object_or_404( - Pharmacy.objects.select_related( + accessible_pharmacies(request.user).select_related( 'organization', 'location_region', 'head_pharmacist__post', 'responsible_manager', 'curator', ).prefetch_related('phones'), @@ -103,7 +105,7 @@ def pharmacy_update_view(request, pk): if not user_can_edit_pharmacy(request.user, pharmacy): return HttpResponseForbidden() - form = PortalPharmacyEditForm(request.POST, instance=pharmacy) + form = PortalPharmacyEditForm(request.POST, instance=pharmacy, user=request.user) if form.is_valid(): pharmacy = form.save(commit=False) pharmacy._history_user = request.user diff --git a/apps/nextcloud_connector/tests.py b/apps/nextcloud_connector/tests.py index 0ba2a4e..5bea9f9 100644 --- a/apps/nextcloud_connector/tests.py +++ b/apps/nextcloud_connector/tests.py @@ -127,7 +127,10 @@ class AccessControlTests(TestCase): self.assertTrue(can_access_pharmacy_docs(self.emp_a, self.ph_a)) self.assertFalse(can_access_pharmacy_docs(self.emp_a, self.ph_b)) - def test_staff_sees_all(self): + def test_staff_needs_explicit_network_access(self): + from core.test_utils import grant_all_locations + self.assertFalse(can_access_pharmacy_docs(self.staff, self.ph_a)) + grant_all_locations(self.staff) self.assertTrue(can_access_pharmacy_docs(self.staff, self.ph_a)) self.assertTrue(can_access_pharmacy_docs(self.staff, self.ph_b)) @@ -160,7 +163,7 @@ class DocumentsViewTests(TestCase): def test_other_pharmacy_forbidden(self): self.client.force_login(self.emp_a) r = self.client.get(f'/pharmacies/{self.ph_b.pk}/documents/') - self.assertEqual(r.status_code, 403) + self.assertEqual(r.status_code, 404) def test_anonymous_redirected(self): r = self.client.get(f'/pharmacies/{self.ph_a.pk}/documents/') @@ -174,7 +177,7 @@ class DocumentsViewTests(TestCase): def test_download_other_pharmacy_forbidden(self): self.client.force_login(self.emp_a) r = self.client.get(f'/pharmacies/{self.ph_b.pk}/documents/download/?name=x.pdf') - self.assertEqual(r.status_code, 403) + self.assertEqual(r.status_code, 404) from apps.nextcloud_connector.access import ( diff --git a/apps/nextcloud_connector/views.py b/apps/nextcloud_connector/views.py index e4ba408..6f9540f 100644 --- a/apps/nextcloud_connector/views.py +++ b/apps/nextcloud_connector/views.py @@ -50,7 +50,8 @@ MAX_UPLOAD_BYTES = 100 * 1024 * 1024 # 100 МБ def _get_pharmacy_or_403(request, pk): """Возвращает (pharmacy, None) или (None, HttpResponseForbidden).""" - pharmacy = get_object_or_404(Pharmacy, pk=pk) + from core.location_access import accessible_pharmacies + pharmacy = get_object_or_404(accessible_pharmacies(request.user), pk=pk) if not can_access_pharmacy_docs(request.user, pharmacy): return None, HttpResponseForbidden('Нет доступа к документам этой аптеки.') return pharmacy, None @@ -508,8 +509,13 @@ def _resolve_token_relpath(payload: dict) -> tuple[str, str, bool] | None: if not name or not is_office_file(name): return None if kind == 'pharmacy': - pharmacy = Pharmacy.objects.filter(pk=payload.get('pharmacy_id')).first() - if pharmacy is None: + from django.contrib.auth import get_user_model + from core.location_access import accessible_pharmacies + user = get_user_model().objects.filter(pk=payload.get('uid'), is_active=True).first() + if user is None: + return None + pharmacy = accessible_pharmacies(user).filter(pk=payload.get('pharmacy_id')).first() + if pharmacy is None or not can_access_pharmacy_docs(user, pharmacy): return None return f'{pharmacy_folder_relpath(pharmacy)}/{name}', name, True if kind == 'report_doc': diff --git a/apps/power_assets/tests.py b/apps/power_assets/tests.py index d2e45b9..676c887 100644 --- a/apps/power_assets/tests.py +++ b/apps/power_assets/tests.py @@ -4,13 +4,13 @@ from django.test import Client, TestCase from apps.locations.models.base import Location from apps.power_assets.models import PowerAsset, PowerAssetService from core.models.directories import Directory, DirectoryType -from core.test_utils import login_2fa +from core.test_utils import login_2fa, grant_all_locations from core.tests_portal_access import make_employee class PowerAssetsTests(TestCase): def setUp(self): - self.user = make_employee(is_staff=True) + self.user = grant_all_locations(make_employee(is_staff=True)) perm = Permission.objects.get(codename='access_utilities') self.user.user_permissions.add(perm) self.client = Client() diff --git a/apps/power_assets/views.py b/apps/power_assets/views.py index 36b054c..7172b46 100644 --- a/apps/power_assets/views.py +++ b/apps/power_assets/views.py @@ -13,6 +13,8 @@ from core.models.directories import Directory from core.portal_access import user_can_access from core.utils.pagination import paginate +from core.location_access import accessible_locations, scoped_queryset, require_location + from .models import PowerAsset, PowerAssetService @@ -20,9 +22,9 @@ def _can_manage_power(user) -> bool: return user_can_access(user, 'utilities') -def _locations_qs(): +def _locations_qs(user): return ( - Location.objects.filter(is_active=True) + accessible_locations(user).filter(is_active=True) .select_related('location_region') .prefetch_related( models.Prefetch( @@ -49,7 +51,7 @@ def _power_assets_context(request, **extra): due_only = (request.GET.get('power_due') or '').strip() == '1' qs = ( - PowerAsset.objects.select_related( + scoped_queryset(request.user, PowerAsset.objects.all()).select_related( 'asset_type', 'status', 'location', 'location__location_region', ) .prefetch_related( @@ -72,7 +74,7 @@ def _power_assets_context(request, **extra): page = paginate(request, qs, page_param='power_page', size_param='power_per_page') - overdue_count = PowerAsset.objects.filter( + overdue_count = scoped_queryset(request.user, PowerAsset.objects.all()).filter( next_service_at__isnull=False, next_service_at__lt=timezone.localdate(), ).count() @@ -82,16 +84,16 @@ def _power_assets_context(request, **extra): 'power_asset_types': _directory_choices('power_asset_type'), 'power_asset_statuses': _directory_choices('power_asset_status'), 'power_service_kinds': PowerAssetService.ServiceKind.choices, - 'power_locations': _locations_qs(), + 'power_locations': _locations_qs(request.user), 'power_type_filter': type_filter, 'power_location_filter': location_filter, 'power_status_filter': status_filter, 'power_due_filter': due_only, 'power_today': timezone.localdate(), 'power_counts': { - 'total': PowerAsset.objects.count(), - 'inverter': PowerAsset.objects.filter(asset_type__code='inverter').count(), - 'generator': PowerAsset.objects.filter(asset_type__code='generator').count(), + 'total': scoped_queryset(request.user, PowerAsset.objects.all()).count(), + 'inverter': scoped_queryset(request.user, PowerAsset.objects.all()).filter(asset_type__code='inverter').count(), + 'generator': scoped_queryset(request.user, PowerAsset.objects.all()).filter(asset_type__code='generator').count(), 'overdue': overdue_count, }, } @@ -107,6 +109,7 @@ def _parse_asset_form(request, asset: PowerAsset | None = None) -> tuple[PowerAs if not asset_type_id or not location_id or not model_name: return None, 'Укажите тип, объект и модель.' + require_location(request.user, location_id) obj = asset or PowerAsset() obj.asset_type_id = asset_type_id obj.location_id = location_id @@ -194,7 +197,7 @@ def power_asset_create_view(request): def power_asset_update_view(request, pk): if not _can_manage_power(request.user): return HttpResponseForbidden() - asset = get_object_or_404(PowerAsset, pk=pk) + asset = get_object_or_404(scoped_queryset(request.user, PowerAsset.objects.all()), pk=pk) obj, err = _parse_asset_form(request, asset) extra = {} if err: @@ -210,7 +213,7 @@ def power_asset_update_view(request, pk): def power_asset_service_create_view(request, pk): if not _can_manage_power(request.user): return HttpResponseForbidden() - asset = get_object_or_404(PowerAsset, pk=pk) + asset = get_object_or_404(scoped_queryset(request.user, PowerAsset.objects.all()), pk=pk) log, err = _parse_service_form(request, asset) extra = {} if err: diff --git a/apps/reports/documents.py b/apps/reports/documents.py index 2c68b69..ac5bac6 100644 --- a/apps/reports/documents.py +++ b/apps/reports/documents.py @@ -13,6 +13,7 @@ from apps.reports.it_monthly.builder import ( from apps.reports.models import ReportDocument from apps.reports.registry import visible_reports from core.portal_access import user_can_access +from core.location_access import has_all_locations REPORTS_DAV_ROOT = '06_Отчёты' ALLOWED_UPLOAD_EXTS = frozenset({'docx', 'xlsx'}) @@ -38,6 +39,8 @@ def user_can_open_reports_hub(user) -> bool: def user_can_view_report_document(user, doc: ReportDocument) -> bool: + if doc.kind == ReportDocument.KIND_SNAPSHOT and not has_all_locations(user): + return False if not user_can_open_reports_hub(user): return False # Черновики — только руководителю своего отдела. @@ -52,6 +55,8 @@ def user_can_view_report_document(user, doc: ReportDocument) -> bool: def user_can_edit_report_document(user, doc: ReportDocument) -> bool: + if doc.kind == ReportDocument.KIND_SNAPSHOT and not has_all_locations(user): + return False if not user_can_open_reports_hub(user): return False if user_can_manage_it_monthly_report(user) and getattr(user, 'department_id', None) == doc.department_id: @@ -79,6 +84,8 @@ def visible_report_documents(user): if not user_can_open_reports_hub(user): return ReportDocument.objects.none() qs = ReportDocument.objects.select_related('department', 'created_by', 'published_by') + if not has_all_locations(user): + qs = qs.exclude(kind=ReportDocument.KIND_SNAPSHOT) dept_id = getattr(user, 'department_id', None) manage_own = user_can_manage_it_monthly_report(user) and bool(dept_id) diff --git a/apps/reports/it_monthly/builder.py b/apps/reports/it_monthly/builder.py index 7a74571..7828985 100644 --- a/apps/reports/it_monthly/builder.py +++ b/apps/reports/it_monthly/builder.py @@ -10,6 +10,7 @@ from django.contrib.auth import get_user_model from django.utils import timezone from apps.it_manager.models import WorkLog +from core.location_access import has_all_locations from apps.it_manager.report_import import default_report_tool_root, load_company, normalize_name from apps.locations.models.base import pharmacy_object_code from apps.reports.it_monthly.conclusions import build_conclusions @@ -90,6 +91,8 @@ def user_can_view_all_department_reports(user) -> bool: def user_can_view_it_monthly_report(user) -> bool: """Карточка IT-отчёта: руководитель; view-all; сотрудники отдела после публикации.""" + if not has_all_locations(user): + return False if user_can_manage_it_monthly_report(user): return True if user_can_view_all_department_reports(user): diff --git a/apps/reports/tests.py b/apps/reports/tests.py index 9fbfdab..cd7c489 100644 --- a/apps/reports/tests.py +++ b/apps/reports/tests.py @@ -15,7 +15,7 @@ from apps.reports.it_monthly.builder import split_work_items from apps.reports.it_monthly.conclusions import build_conclusions from apps.reports.registry import visible_reports from core.models.directories import Directory, DirectoryType -from core.test_utils import login_2fa +from core.test_utils import login_2fa, grant_all_locations class MonthlyReportTests(TestCase): @@ -36,7 +36,9 @@ class MonthlyReportTests(TestCase): ) it_group = Group.objects.get(name='IT') self.head.groups.add(it_group) + grant_all_locations(self.head) self.it_user.groups.add(it_group) + grant_all_locations(self.it_user) self.client = Client() region_type, _ = DirectoryType.objects.get_or_create(code='region', defaults={'name': 'Регион'}) @@ -472,7 +474,9 @@ class ReportTemplateTests(TestCase): ) it_group = Group.objects.get(name='IT') self.head.groups.add(it_group) + grant_all_locations(self.head) self.it_user.groups.add(it_group) + grant_all_locations(self.it_user) self.client = Client() self.classic, _ = ReportTemplate.objects.update_or_create( @@ -658,7 +662,9 @@ class ReportDocumentTests(TestCase): ) it_group = Group.objects.get(name='IT') self.head.groups.add(it_group) + grant_all_locations(self.head) self.member.groups.add(it_group) + grant_all_locations(self.member) self.client = Client() def test_hub_shows_documents_block_for_head(self): @@ -757,6 +763,7 @@ class ReportDocumentTests(TestCase): ) it_group = Group.objects.get(name='IT') viewer.groups.add(it_group) + grant_all_locations(viewer) perm = Permission.objects.get( codename='view_all_departments', content_type__app_label='reports', ) diff --git a/apps/reports/views.py b/apps/reports/views.py index e8a5ffb..8b0d1c1 100644 --- a/apps/reports/views.py +++ b/apps/reports/views.py @@ -65,8 +65,19 @@ def _onlyoffice_ready() -> bool: return onlyoffice_is_ready() +def _authorized_monthly_report(user, year, month): + from core.location_access import has_all_locations + from django.core.exceptions import PermissionDenied + if not has_all_locations(user): + raise PermissionDenied('Сводный отчёт требует доступа ко всей сети.') + return build_monthly_report(year, month) + + def _resolve_it_month(request, month_raw: str | None) -> tuple[int, int] | HttpResponse: """Выбрать месяц с учётом публикации; 403 если нет доступа к периоду.""" + from core.location_access import has_all_locations + if not has_all_locations(request.user): + return HttpResponseForbidden() months = available_months_for_user(request.user) year, month = parse_month_id(month_raw) mid = f'{year:04d}-{month:02d}' @@ -133,7 +144,7 @@ def _it_monthly_page_context(request, year: int, month: int) -> dict: 'onlyoffice_ready': _onlyoffice_ready(), } if show_live: - report = build_monthly_report(year, month) + report = _authorized_monthly_report(request.user, year, month) report['available_months'] = available_months_for_user(request.user) or [report['id']] context['report'] = report else: @@ -200,7 +211,7 @@ def _export_with_template(request, month_id, fmt: str): year, month = parse_month_id(month_id) if not user_can_access_it_monthly_month(request.user, year, month): return HttpResponseForbidden() - report = build_monthly_report(year, month) + report = _authorized_monthly_report(request.user, year, month) tpl = _default_it_template() try: payload = render_report(report, tpl, fmt) @@ -247,7 +258,7 @@ def it_monthly_preview_view(request, month_id): year, month = parse_month_id(month_id) if not user_can_access_it_monthly_month(request.user, year, month): return HttpResponseForbidden() - report = build_monthly_report(year, month) + report = _authorized_monthly_report(request.user, year, month) tpl = _default_it_template() try: html = render_report(report, tpl, 'html') @@ -266,7 +277,7 @@ def _prepare_it_monthly_word(request, year: int, month: int) -> HttpResponse: return HttpResponseForbidden() if not user_can_manage_it_monthly_report(request.user): return HttpResponseForbidden() - report = build_monthly_report(year, month) + report = _authorized_monthly_report(request.user, year, month) tpl = _default_it_template() try: payload = render_report(report, tpl, 'docx') @@ -635,7 +646,7 @@ def template_preview_draft_view(request, pk): pass year, month = parse_month_id(request.POST.get('month')) - report = build_monthly_report(year, month) + report = _authorized_monthly_report(request.user, year, month) ctx = build_template_context( report, company=load_default_company(), labels=blocks.get('labels'), ) diff --git a/apps/rro/access.py b/apps/rro/access.py index 20f8a17..49b7f7b 100644 --- a/apps/rro/access.py +++ b/apps/rro/access.py @@ -1,44 +1,6 @@ -"""Доступ к реестру РРО: полный (офис/IT) или только своя аптека (заведующий).""" -from __future__ import annotations - -from typing import Literal - -from apps.locations.models import Pharmacy +"""Права на операции РРО; область данных — core.location_access.""" from core.portal_access import user_can_access -MANAGER_GROUP = 'Заведующий аптекой' -FULL_REGISTRY_GROUPS = frozenset({ - 'IT', - 'Ревизор', - 'Администратор сети', -}) - -RroScope = Literal['all'] | Pharmacy | None - - -def is_pharmacy_scoped_rro_user(user) -> bool: - """Заведующий аптекой — только записи своей аптеки (без полного реестра).""" - if not user or not getattr(user, 'is_authenticated', False): - return False - if getattr(user, 'is_superuser', False): - return False - if not user.groups.filter(name=MANAGER_GROUP).exists(): - return False - if user.groups.filter(name__in=FULL_REGISTRY_GROUPS).exists(): - return False - return True - - -def rro_list_scope(user) -> RroScope: - """'all' — полный реестр; Pharmacy — одна аптека; None — нет привязки.""" - if not user_can_access(user, 'rro'): - return None - if getattr(user, 'is_superuser', False): - return 'all' - if is_pharmacy_scoped_rro_user(user): - return getattr(user, 'user_pharmacy', None) - return 'all' - def can_edit_rro_registers(user) -> bool: """Создание/редактирование — только staff (офис/IT), не заведующие.""" @@ -47,9 +9,5 @@ def can_edit_rro_registers(user) -> bool: def user_can_access_register(user, register) -> bool: """Проверка доступа к конкретной записи (для update).""" - scope = rro_list_scope(user) - if scope == 'all': - return True - if scope is None: - return False - return register.pharmacy_id == scope.pk + from core.location_access import accessible_pharmacies + return user_can_access(user, 'rro') and accessible_pharmacies(user).filter(pk=register.pharmacy_id).exists() diff --git a/apps/rro/tests.py b/apps/rro/tests.py index 3531a5d..88b1f32 100644 --- a/apps/rro/tests.py +++ b/apps/rro/tests.py @@ -1,3 +1,4 @@ +from core.test_utils import grant_all_locations from datetime import date from django.contrib.auth import get_user_model @@ -137,6 +138,7 @@ class CashRegisterPortalTests(TestCase): ad_sync_enabled=False, totp_enabled=True, ) + grant_all_locations(cls.user) grant_rro_access(cls.user) def _login(self): @@ -312,6 +314,7 @@ class CashRegisterCrudTests(TestCase): mattermost_sync_enabled=False, ad_sync_enabled=False, ) + grant_all_locations(cls.staff) grant_rro_access(cls.staff) def _login(self, user): diff --git a/apps/rro/views.py b/apps/rro/views.py index 1243806..b6c7295 100644 --- a/apps/rro/views.py +++ b/apps/rro/views.py @@ -15,10 +15,11 @@ from core.models.directories import Directory from core.portal_access import section_required from core.utils.htmx import render_page +from core.location_access import accessible_pharmacies, scoped_queryset, has_all_locations + from .models import CashRegister from .access import ( can_edit_rro_registers, - rro_list_scope, user_can_access_register, ) @@ -124,11 +125,10 @@ def _parse_register_form(post, *, instance=None): def _rro_list_context(request, **extra): user = request.user - scope = rro_list_scope(user) can_edit = can_edit_rro_registers(user) - full_registry = scope == 'all' + full_registry = True # фильтры доступны и для нескольких закреплённых аптек - registers = CashRegister.objects.select_related( + registers = scoped_queryset(user, CashRegister.objects.all(), 'pharmacy').select_related( 'pharmacy', 'pharmacy__organization', 'vendor', 'status', ) @@ -144,12 +144,6 @@ def _rro_list_context(request, **extra): registers = registers.filter(pharmacy_id=filter_pharmacy) if filter_org: registers = registers.filter(pharmacy__organization_id=filter_org) - elif scope is not None: - registers = registers.filter(pharmacy=scope) - filter_pharmacy = str(scope.pk) - filter_org = '' - else: - registers = registers.none() if filter_kind in {CashRegister.Kind.RRO, CashRegister.Kind.PRRO}: registers = registers.filter(kind=filter_kind) @@ -194,7 +188,7 @@ def _rro_list_context(request, **extra): 'rro_full_registry': full_registry, 'is_staff': full_registry, # совместимость шаблона 'can_edit_rro': can_edit, - 'rro_pharmacy_scoped': scope is not None and scope != 'all', + 'rro_pharmacy_scoped': not has_all_locations(user), 'registers': register_list, 'registers_json': json.dumps(registers_js, cls=DjangoJSONEncoder, ensure_ascii=False), 'statuses': statuses, @@ -208,13 +202,13 @@ def _rro_list_context(request, **extra): 'show_create_panel': show_create and can_edit, 'prefill_pharmacy': filter_pharmacy if can_edit else '', 'all_pharmacies': ( - Pharmacy.objects.filter(is_active=True) + accessible_pharmacies(user).filter(is_active=True) .select_related('organization') .order_by('organization__name', 'pharmacy_number') if full_registry else [] ), 'organizations': ( - Directory.objects.filter(directory_type__code='organization', is_active=True).order_by('name') + Directory.objects.filter(pk__in=accessible_pharmacies(user).values('organization_id'), is_active=True).order_by('name') if full_registry else [] ), } @@ -251,8 +245,7 @@ def rro_create_view(request): form_data = dict(request.POST.items()) try: register, form_data = _parse_register_form(request.POST) - scope = rro_list_scope(request.user) - if scope is not None and scope != 'all' and register.pharmacy_id != scope.pk: + if not user_can_access_register(request.user, register): return HttpResponseForbidden() with transaction.atomic(): register.save() @@ -280,7 +273,7 @@ def rro_update_view(request, pk): if not can_edit_rro_registers(request.user): return HttpResponseForbidden() - register = get_object_or_404(CashRegister, pk=pk) + register = get_object_or_404(scoped_queryset(request.user, CashRegister.objects.all(), 'pharmacy'), pk=pk) if not user_can_access_register(request.user, register): return HttpResponseForbidden() form_data = dict(request.POST.items()) diff --git a/apps/utilities/views.py b/apps/utilities/views.py index 397c34a..7a9a738 100644 --- a/apps/utilities/views.py +++ b/apps/utilities/views.py @@ -7,6 +7,8 @@ from apps.locations.models.pharmacies import Pharmacy from core.models.directories import Directory from core.portal_access import section_required +from core.location_access import accessible_pharmacies, scoped_queryset, require_location + from .helpers import ( attach_reading_display, get_reading_mode, @@ -25,6 +27,7 @@ def _utilities_context(request, **extra): """Показання лічильників: staff бачить усі аптеки й фільтри.""" is_staff = request.user.is_staff own_pharmacy = request.user.user_pharmacy + pharmacies = accessible_pharmacies(request.user).filter(is_active=True) form_error = '' config_message = '' @@ -37,6 +40,7 @@ def _utilities_context(request, **extra): utility_type_id = request.POST.get('config_utility_type') reading_mode = request.POST.get('reading_mode', UtilityMeterConfig.READING_MODE_SINGLE) if pharmacy_id and utility_type_id and reading_mode in dict(UtilityMeterConfig.READING_MODE_CHOICES): + require_location(request.user, pharmacy_id) config, _created = UtilityMeterConfig.objects.update_or_create( pharmacy_id=pharmacy_id, utility_type_id=utility_type_id, @@ -48,7 +52,7 @@ def _utilities_context(request, **extra): else: config_message = 'Заповніть аптеку, тип лічильника та режим.' else: - pharmacy_id = request.POST.get('pharmacy') if is_staff else (own_pharmacy.pk if own_pharmacy else None) + pharmacy_id = request.POST.get('pharmacy') or (own_pharmacy.pk if own_pharmacy else None) utility_type_id = request.POST.get('utility_type') reading_date = request.POST.get('reading_date') value = request.POST.get('value') @@ -60,6 +64,7 @@ def _utilities_context(request, **extra): elif not (utility_type_id and reading_date): form_error = 'Показання не збережено: заповніть тип лічильника та дату.' else: + require_location(request.user, pharmacy_id) mode = get_reading_mode(pharmacy_id, utility_type_id) reading = UtilityReading( pharmacy_id=pharmacy_id, @@ -81,24 +86,16 @@ def _utilities_context(request, **extra): utility_types = Directory.objects.filter(directory_type__code='utility_type', is_active=True).order_by('name') - readings = UtilityReading.objects.select_related( + readings = scoped_queryset(request.user, UtilityReading.objects.all(), 'pharmacy').select_related( 'pharmacy', 'pharmacy__organization', 'utility_type', 'submitted_by', ) - if is_staff: - filter_pharmacy = request.GET.get('pharmacy', '') - filter_type = request.GET.get('utility_type', '') - if filter_pharmacy: - readings = readings.filter(pharmacy_id=filter_pharmacy) - if filter_type: - readings = readings.filter(utility_type_id=filter_type) - pharmacies_for_form = Pharmacy.objects.filter(is_active=True).select_related('organization').order_by('organization__name', 'pharmacy_number') - else: - filter_pharmacy = filter_type = '' - if own_pharmacy: - readings = readings.filter(pharmacy=own_pharmacy) - else: - readings = readings.none() - pharmacies_for_form = [own_pharmacy] if own_pharmacy else [] + filter_pharmacy = request.GET.get('pharmacy', '') + filter_type = request.GET.get('utility_type', '') + if filter_pharmacy: + readings = readings.filter(pharmacy_id=filter_pharmacy) + if filter_type: + readings = readings.filter(utility_type_id=filter_type) + pharmacies_for_form = pharmacies.select_related('organization').order_by('organization__name', 'pharmacy_number') readings = list(readings[:200]) attach_reading_display(readings) @@ -106,12 +103,10 @@ def _utilities_context(request, **extra): r.can_delete = can_delete_reading(request.user, r) meter_configs = list( - UtilityMeterConfig.objects.select_related('pharmacy', 'pharmacy__organization', 'utility_type') + scoped_queryset(request.user, UtilityMeterConfig.objects.all(), 'pharmacy').select_related('pharmacy', 'pharmacy__organization', 'utility_type') .filter(is_active=True) .order_by('pharmacy__organization__name', 'pharmacy__pharmacy_number', 'utility_type__name') ) - if not is_staff and own_pharmacy: - meter_configs = [c for c in meter_configs if c.pharmacy_id == own_pharmacy.pk] import json from django.core.serializers.json import DjangoJSONEncoder @@ -126,9 +121,9 @@ def _utilities_context(request, **extra): 'config_message': config_message, 'filter_pharmacy': filter_pharmacy, 'filter_type': filter_type, - 'all_pharmacies': Pharmacy.objects.filter(is_active=True).select_related('organization').order_by('organization__name', 'pharmacy_number') if is_staff else [], + 'all_pharmacies': pharmacies.select_related('organization').order_by('organization__name', 'pharmacy_number'), 'meter_configs': meter_configs, - 'meter_configs_json': json.dumps(meter_configs_map(), cls=DjangoJSONEncoder), + 'meter_configs_json': json.dumps({f'{c.pharmacy_id}:{c.utility_type_id}': c.reading_mode for c in meter_configs}, cls=DjangoJSONEncoder), 'utility_type_modes_json': json.dumps(utility_type_modes_map(utility_types), cls=DjangoJSONEncoder), 'reading_mode_choices': UtilityMeterConfig.READING_MODE_CHOICES, 'own_pharmacy_id': own_pharmacy.pk if own_pharmacy else '', @@ -150,7 +145,7 @@ def utilities_view(request): def utility_reading_delete_view(request, pk): from core.accounting_portal import render_accounting_portal - reading = get_object_or_404(UtilityReading, pk=pk) + reading = get_object_or_404(scoped_queryset(request.user, UtilityReading.objects.all(), 'pharmacy'), pk=pk) if can_delete_reading(request.user, reading): reading.delete() return render_accounting_portal(request, 'utilities') @@ -164,6 +159,6 @@ def utility_meter_config_delete_view(request, pk): if not request.user.is_staff: return render_accounting_portal(request, 'utilities', config_message='Немає прав.') - config = get_object_or_404(UtilityMeterConfig, pk=pk) + config = get_object_or_404(scoped_queryset(request.user, UtilityMeterConfig.objects.all(), 'pharmacy'), pk=pk) config.delete() return render_accounting_portal(request, 'utilities', config_message='Налаштування видалено.') diff --git a/core/dashboard.py b/core/dashboard.py index 213a084..2d52f7b 100644 --- a/core/dashboard.py +++ b/core/dashboard.py @@ -34,17 +34,34 @@ def _ctx_metrics(request) -> dict: from core.models import Directory, DirectoryType from core.middleware import ONLINE_WINDOW - active = Employee.objects.filter(is_active=True) - return { - 'employees_online': active.filter( - last_seen__gte=timezone.now() - ONLINE_WINDOW).count(), - 'employees_active': active.count(), - 'pharmacies_active': Pharmacy.objects.filter(is_active=True).count(), - 'open_tickets': Ticket.objects.filter( - status__code__in=['new', 'in_progress', 'waiting']).count(), - 'directory_types_count': DirectoryType.objects.filter(is_active=True).count(), - 'directories_count': Directory.objects.count(), - } + from core.portal_access import user_can_access + from core.location_access import accessible_pharmacies, scoped_employees, scoped_queryset + + # Каждая карточка проверяет свой раздел: reports не даёт доступ к IT + # или справочникам. Не запрашиваем закрытые данные даже для контекста. + context = {} + if user_can_access(request.user, 'hr_portal'): + active = scoped_employees(request.user, Employee.objects.filter(is_active=True)) + context.update( + employees_online=active.filter( + last_seen__gte=timezone.now() - ONLINE_WINDOW).count(), + employees_active=active.count(), + ) + if user_can_access(request.user, 'pharmacies'): + context['pharmacies_active'] = accessible_pharmacies(request.user).filter(is_active=True).count() + tickets = Ticket.objects.filter(status__code__in=['new', 'in_progress', 'waiting']) + all_tickets = user_can_access(request.user, 'it_manager') + if all_tickets: + tickets = scoped_queryset(request.user, tickets) + else: + tickets = tickets.filter(requester=request.user) + context.update(open_tickets=tickets.count(), tickets_are_personal=not all_tickets) + if user_can_access(request.user, 'databases'): + context.update( + directory_types_count=DirectoryType.objects.filter(is_active=True).count(), + directories_count=Directory.objects.count(), + ) + return context def _ctx_news(request) -> dict: @@ -119,27 +136,27 @@ DASHBOARD_WIDGETS: tuple[DashboardWidget, ...] = ( DashboardWidget( 'my_tasks', 'Мои задачи', 'Мои активные задачи (где я автор, ответственный, соисполнитель).', - 'my_tasks', _ctx_my_tasks, + 'my_tasks', _ctx_my_tasks, required_section='tasks', ), DashboardWidget( 'upcoming', 'Ближайшие события', 'Праздники, отпуска, дни рождения, смены и встречи на 2 недели вперёд.', - 'upcoming', _ctx_upcoming, + 'upcoming', _ctx_upcoming, required_section='calendar', ), DashboardWidget( 'my_hr_requests', 'Мои кадровые заявки', 'Статусы моих заявок на отпуск, больничный, отгул и т.п.', - 'my_hr_requests', _ctx_my_hr_requests, + 'my_hr_requests', _ctx_my_hr_requests, required_section='hr_portal', ), DashboardWidget( 'my_tickets', 'Мои IT-заявки', 'Мои открытые заявки в IT-поддержку и их статус.', - 'my_tickets', _ctx_my_tickets, + 'my_tickets', _ctx_my_tickets, required_section='it_support', ), DashboardWidget( 'news', 'Последние новости', 'Свежие публикации и объявления компании.', - 'news', _ctx_news, + 'news', _ctx_news, required_section='news', ), DashboardWidget( 'quick_access', 'Быстрый доступ', @@ -158,8 +175,10 @@ def valid_keys(keys) -> list: """Только существующие ключи, без дублей, с сохранением порядка.""" seen = set() result = [] - for k in keys or []: - if k in WIDGETS_BY_KEY and k not in seen: + if not isinstance(keys, (list, tuple)): + return [] + for k in keys: + if isinstance(k, str) and k in WIDGETS_BY_KEY and k not in seen: seen.add(k) result.append(k) return result @@ -167,6 +186,8 @@ def valid_keys(keys) -> list: def widget_allowed(widget: DashboardWidget, user) -> bool: """Доступен ли виджет пользователю (по разделу портала).""" + if not user or not getattr(user, 'is_authenticated', False) or not user.is_active: + return False if widget.required_section is None: return True from core.portal_access import user_can_access @@ -194,10 +215,8 @@ def resolve_layout_keys(user) -> list: allowed = set(accessible_keys(user)) layout = DashboardLayout.objects.filter(employee=user).first() - if layout is not None and layout.widgets: - keys = [k for k in valid_keys(layout.widgets) if k in allowed] - if keys: - return keys + if layout is not None: + return [k for k in valid_keys(layout.widgets) if k in allowed] default = default_keys_for(getattr(user, 'user_location_type', None)) return [k for k in default if k in allowed] @@ -224,8 +243,7 @@ def build_dashboard_context(request) -> dict: def _has_custom_layout(user) -> bool: from core.models import DashboardLayout - layout = DashboardLayout.objects.filter(employee=user).first() - return bool(layout and layout.widgets) + return DashboardLayout.objects.filter(employee=user).exists() def customize_rows(user) -> list: diff --git a/core/test_utils.py b/core/test_utils.py index acfaa26..0322857 100644 --- a/core/test_utils.py +++ b/core/test_utils.py @@ -16,3 +16,14 @@ def login_past_2fa(client, user): # Обратная совместимость с существующими тестами. login_2fa = login_past_2fa + + +def grant_all_locations(user): + """Явная область всей сети для тестов центральных операторов.""" + from django.contrib.auth.models import Permission + user.user_permissions.add(Permission.objects.get( + content_type__app_label='locations', codename='view_all_locations', + )) + for key in ('_perm_cache', '_user_perm_cache', '_group_perm_cache'): + user.__dict__.pop(key, None) + return user diff --git a/core/tests_dashboard.py b/core/tests_dashboard.py index 9ff7219..90dea78 100644 --- a/core/tests_dashboard.py +++ b/core/tests_dashboard.py @@ -185,3 +185,131 @@ class DashboardViewTests(TestCase): ) self.assertEqual(resp.status_code, 302) self.assertFalse(DashboardLayout.objects.filter(employee=user).exists()) + + +class DashboardSecurityRegressionTests(TestCase): + def request_for(self, user): + from django.test import RequestFactory + request = RequestFactory().get('/dashboard/') + request.user = user + return request + + def test_no_group_does_not_build_closed_widget_contexts(self): + from unittest.mock import patch + user = make_employee() + user.groups.clear() + DashboardLayout.objects.create(employee=user, widgets=list(dash.DEFAULT_WIDGET_KEYS)) + with patch('apps.visit_manager.calendar_service.upcoming_events') as events: + context = dash.build_dashboard_context(self.request_for(user)) + events.assert_not_called() + self.assertEqual([w['key'] for w in context['dashboard_widgets']], + ['welcome', 'my_tickets', 'quick_access']) + for key in ('latest_posts', 'my_tasks', 'my_hr_requests', 'upcoming_events'): + self.assertNotIn(key, context) + + def test_forged_post_cannot_enable_closed_widgets_or_change_other_layout(self): + user = make_employee() + user.groups.clear() + other = make_employee() + DashboardLayout.objects.create(employee=other, widgets=['news']) + login_2fa(self.client, user) + response = self.client.post('/dashboard/customize/', { + 'employee': other.pk, 'order': ','.join(dash.DEFAULT_WIDGET_KEYS), + 'enabled': dash.DEFAULT_WIDGET_KEYS, + }) + self.assertEqual(response.status_code, 302) + self.assertEqual(DashboardLayout.objects.get(employee=user).widgets, + ['welcome', 'my_tickets', 'quick_access']) + self.assertEqual(DashboardLayout.objects.get(employee=other).widgets, ['news']) + + def test_empty_layout_stays_empty_until_reset(self): + user = make_employee() + dash.save_layout(user, [], []) + context = dash.build_dashboard_context(self.request_for(user)) + self.assertEqual(context['dashboard_widgets'], []) + self.assertTrue(context['dashboard_has_custom_layout']) + login_2fa(self.client, user) + self.assertContains(self.client.get('/dashboard/', HTTP_HX_REQUEST='true'), + 'Все виджеты скрыты.') + + def test_revocation_does_not_restore_default_widgets(self): + user = grant_reports(make_employee()) + dash.save_layout(user, ['metrics'], ['metrics']) + user.groups.clear() + self.assertEqual(dash.resolve_layout_keys(Employee.objects.get(pk=user.pk)), []) + + def test_welcome_buttons_and_widget_links_respect_sections(self): + user = make_employee() + user.groups.clear() + login_2fa(self.client, user) + for htmx in ('true', ''): + response = self.client.get('/dashboard/', HTTP_HX_REQUEST=htmx) + self.assertEqual(response.status_code, 200) + for path in ('/tasks/', '/chat/', '/calendar/', '/news/', '/hr-portal/'): + self.assertNotContains(response, f'hx-get="{path}"') + + def test_anonymous_cannot_read_or_save_dashboard(self): + for path in ('/dashboard/', '/dashboard/customize/'): + self.assertEqual(self.client.get(path).status_code, 302) + self.assertEqual(self.client.post(path, {'enabled': ['news']}).status_code, 302) + self.assertFalse(DashboardLayout.objects.exists()) + + def test_post_requires_csrf(self): + from django.test import Client + client = Client(enforce_csrf_checks=True) + login_2fa(client, make_employee()) + self.assertEqual(client.post('/dashboard/customize/', {'enabled': ['news']}).status_code, 403) + self.assertFalse(DashboardLayout.objects.exists()) + + def test_reports_does_not_grant_other_sections_metrics(self): + user = grant_reports(make_employee(is_staff=True)) + context = dash._ctx_metrics(self.request_for(user)) + for key in ('pharmacies_active', 'directory_types_count', 'directories_count'): + self.assertNotIn(key, context) + self.assertTrue(context['tickets_are_personal']) + login_2fa(self.client, user) + response = self.client.get('/dashboard/', HTTP_HX_REQUEST='true') + self.assertNotContains(response, 'Доступных аптек') + self.assertNotContains(response, 'Справочников') + self.assertContains(response, 'Моих открытых IT-заявок') + + def test_ticket_counts_are_personal_unless_it_access_granted(self): + from apps.it_manager.models import Ticket + from core.models import Directory, DirectoryType + from core.tests_portal_access import portal_perm + user = grant_reports(make_employee()) + other = make_employee() + values = {} + for field, code in [('ticket_type', 'other'), ('priority', 'medium'), ('status', 'new')]: + kind = {'ticket_type': 'ticket_type', 'priority': 'ticket_priority', 'status': 'ticket_status'}[field] + dtype, _ = DirectoryType.objects.get_or_create(code=kind, defaults={'name': kind}) + values[field], _ = Directory.objects.get_or_create( + directory_type=dtype, code=code, defaults={'name': code}) + for employee in (user, other): + Ticket.objects.create(requester=employee, title='Private ticket', description='test', **values) + self.assertEqual(dash._ctx_metrics(self.request_for(user))['open_tickets'], 1) + self.assertEqual(len(dash._ctx_my_tickets(self.request_for(user))['my_tickets']), 1) + user.user_permissions.add(portal_perm('access_it_manager')) + from core.test_utils import grant_all_locations + grant_all_locations(user) + user = Employee.objects.get(pk=user.pk) + context = dash._ctx_metrics(self.request_for(user)) + self.assertEqual(context['open_tickets'], 2) + self.assertFalse(context['tickets_are_personal']) + + def test_tasks_do_not_include_nonparticipant(self): + from apps.tasks.models import Task + user, other = make_employee(), make_employee() + own = Task.objects.create(creator=user, title='Own') + Task.objects.create(creator=other, title='Private') + self.assertEqual(list(dash._ctx_my_tasks(self.request_for(user))['my_tasks']), [own]) + + def test_preview_does_not_inherit_superuser_widgets(self): + user = make_employee(is_superuser=True) + user._access_preview_perms = set() + user._access_preview_groups = [] + self.assertEqual(dash.accessible_keys(user), ['welcome', 'my_tickets', 'quick_access']) + + def test_invalid_saved_json_keys_are_ignored(self): + self.assertEqual(dash.valid_keys([{}, ['news'], 'news', None]), ['news']) + self.assertEqual(dash.valid_keys('news'), []) diff --git a/core/views.py b/core/views.py index 856df5b..49e860f 100644 --- a/core/views.py +++ b/core/views.py @@ -1,3 +1,4 @@ +from core.location_access import accessible_pharmacies, scoped_queryset, has_all_locations from django.core.exceptions import ValidationError from django.db.models import Q, Count, Prefetch, Exists, OuterRef from django.http import HttpResponse, HttpResponseForbidden, JsonResponse, HttpResponseNotAllowed @@ -273,7 +274,7 @@ def build_region_tree(region_nodes): @login_required @section_required('pharmacies') def pharmacies_view(request): - pharmacies = Pharmacy.objects.select_related('organization', 'location_region', 'head_pharmacist').order_by('organization__name', 'pharmacy_number') + pharmacies = accessible_pharmacies(request.user).select_related('organization', 'location_region', 'head_pharmacist').order_by('organization__name', 'pharmacy_number') active_count = pharmacies.filter(is_active=True).count() # Get unique regions for filter @@ -346,7 +347,7 @@ def dislocation_view(request): ) pharmacies = ( - Pharmacy.objects + accessible_pharmacies(request.user) .select_related( 'organization', 'location_region', 'head_pharmacist', 'curator', 'responsible_manager', @@ -595,6 +596,9 @@ def _timetrack_context(request): use_pvms = bool(getattr(settings, 'PVMS_SYNC_ENABLED', False)) sot = bool(getattr(settings, 'PVMS_ATTENDANCE_SOT', False)) if use_pvms: + if not has_all_locations(request.user): + from django.core.exceptions import PermissionDenied + raise PermissionDenied('Сводка PVMS требует доступа ко всей сети.') try: from apps.portal_connector.pvms_client import PvmsClient client = PvmsClient() @@ -622,7 +626,7 @@ def _timetrack_context(request): if rows is None: from apps.visit_manager.models import WorkShift - shifts = (WorkShift.objects + shifts = (scoped_queryset(request.user, WorkShift.objects.all()) .filter(date__gte=period_start, date__lte=period_end) .select_related('employee', 'location')) if location_type in ('office', 'pharmacy', 'warehouse'): @@ -648,7 +652,7 @@ def _timetrack_context(request): 'avg': round(hours / r['days'], 1) if r['days'] else 0.0, }) - base = WorkShift.objects.filter(date__gte=period_start, date__lte=period_end) + base = scoped_queryset(request.user, WorkShift.objects.all()).filter(date__gte=period_start, date__lte=period_end) counts = { 'all': base.values('employee_id').distinct().count(), 'office': base.filter(location__location_type='office').values('employee_id').distinct().count(), @@ -860,7 +864,7 @@ def pharmacy_detail_view(request, pk): from apps.locations.views import _pharmacy_detail_context pharmacy = get_object_or_404( - Pharmacy.objects.select_related( + accessible_pharmacies(request.user).select_related( 'organization', 'location_region', 'head_pharmacist__post', 'responsible_manager', 'curator', ).prefetch_related('phones'), @@ -1134,7 +1138,7 @@ def _hr_portal_context(request): 'hr_access_location_types': { a.id: [lt.code for lt in a.location_types.all()] for a in hr_access_systems }, - 'hr_pharmacies': Pharmacy.objects.filter(is_active=True).order_by('pharmacy_number'), + 'hr_pharmacies': accessible_pharmacies(request.user).filter(is_active=True).order_by('pharmacy_number'), 'hr_locations': Location.objects.filter( location_type__in=['office', 'warehouse'], is_active=True, ).order_by('location_type', 'address'), diff --git a/templates/pages/dashboard_widgets/metrics.html b/templates/pages/dashboard_widgets/metrics.html index b383c8e..286e78b 100644 --- a/templates/pages/dashboard_widgets/metrics.html +++ b/templates/pages/dashboard_widgets/metrics.html @@ -1,6 +1,7 @@ {% load i18n %} -
+ {% if directories_count != None %}- {% if is_staff %}Показання лічильників від усіх аптек{% else %}Показання лічильників вашої аптеки{% endif %} + {% if is_staff %}Показання лічильників доступних аптек{% else %}Показання лічильників доступних аптек{% endif %}
@@ -106,7 +106,7 @@