Security hardening: обязательный ключ шифрования, fail-closed токены, права API

- FIELD_ENCRYPTION_KEY: production требует явный ключ (ImproperlyConfigured
  при пустом), local — fallback на SECRET_KEY с предупреждением (crypto.py)
- MATTERMOST_INTERACTIVE_TOKEN: fail-closed токен для интерактивных кнопок
  Mattermost (bot_service task-done URL + проверка в mattermost views)
- Ужесточение прав API (employees/visit_manager) + тесты
  (employees, mattermost_connector, visit_manager)
- .env.example с описанием переменных; правки шаблонов (index/dislocation/pharmacies)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-27 12:23:28 +03:00
parent 9e31b9d620
commit 38c8e35a51
16 changed files with 335 additions and 37 deletions
+18
View File
@@ -0,0 +1,18 @@
# Django
DJANGO_SETTINGS_MODULE=acore.settings.local
DEBUG=True
SECRET_KEY=change-me-in-production
ALLOWED_HOSTS=localhost,127.0.0.1
DATABASE_URL=postgres://user:pass@localhost:5432/acore
# Fernet field encryption for ConnectorConfig / bot tokens / Carbonio passwords.
# MUST be set in production (acore.settings.production refuses empty value).
# Keep equal to the key used when secrets were encrypted, or re-encrypt after rotation.
# Prefer a value DISTINCT from SECRET_KEY once you are ready to rotate.
FIELD_ENCRYPTION_KEY=
# Mattermost integrations (fail-closed when empty)
MATTERMOST_SLASH_COMMAND_TOKEN=
# Interactive button actions; defaults to slash token if unset
MATTERMOST_INTERACTIVE_TOKEN=
PORTAL_BASE_URL=http://localhost:8000
+1
View File
@@ -147,6 +147,7 @@ media/
# Project-specific
.env*
!.env.example
.directory
packets.txt
+10 -3
View File
@@ -12,9 +12,9 @@ ALLOWED_HOSTS = env.list('ALLOWED_HOSTS', default=[])
# Ключ для шифрования секретных полей в БД (EncryptedCharField: пароли/токены
# коннекторов, access-токены Mattermost). Задавайте ОТДЕЛЬНО от SECRET_KEY —
# тогда утечка SECRET_KEY не раскрывает секреты at-rest. При отсутствии
# core.utils.crypto делает fallback на SECRET_KEY (см. предупреждение ниже).
FIELD_ENCRYPTION_KEY = env('FIELD_ENCRYPTION_KEY', default=env('SECRET_KEY'))
# тогда утечка SECRET_KEY не раскрывает секреты at-rest.
# Пустой default: production требует явный ключ; local может fallback на SECRET_KEY.
FIELD_ENCRYPTION_KEY = env('FIELD_ENCRYPTION_KEY', default='')
# E016 — ложноположительный для proxy-моделей над Directory (django-fast-treenode
# динамически добавляет индекс по _path, который proxy «наследует»). На реальных
@@ -30,6 +30,13 @@ PORTAL_BASE_URL = env('PORTAL_BASE_URL', default='http://localhost:8000')
# пока команда реально не создана в Mattermost и токен не прописан в .env.
MATTERMOST_SLASH_COMMAND_TOKEN = env('MATTERMOST_SLASH_COMMAND_TOKEN', default='')
# Токен интерактивных кнопок Mattermost (attachment actions). Fail-closed:
# пустой токен → task_done_action отклоняет все запросы. Можно совпадать со
# slash-токеном или задать отдельный секрет.
MATTERMOST_INTERACTIVE_TOKEN = env(
'MATTERMOST_INTERACTIVE_TOKEN',
default=env('MATTERMOST_SLASH_COMMAND_TOKEN', default=''),
)
INSTALLED_APPS = [
# Admin
"unfold", # before django.contrib.admin
+9
View File
@@ -1,3 +1,5 @@
from django.core.exceptions import ImproperlyConfigured
from .base import * # noqa
DEBUG = False
@@ -7,3 +9,10 @@ SESSION_COOKIE_SECURE = True
SECURE_HSTS_SECONDS = 31536000
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_SSL_REDIRECT = True
# В production ключ шифрования полей обязан быть задан явно (не fallback на SECRET_KEY).
if not FIELD_ENCRYPTION_KEY:
raise ImproperlyConfigured(
'FIELD_ENCRYPTION_KEY must be set in production (.env). '
'Use a value distinct from SECRET_KEY when rotating secrets.'
)
+35
View File
@@ -103,3 +103,38 @@ class EmployeePermissionOverrideTests(TestCase):
self.assertTrue(o.is_currently_active())
o.is_active = False
self.assertFalse(o.is_currently_active())
class EmployeeAPITests(TestCase):
"""EmployeeViewSet — только чтение и только для staff."""
def setUp(self):
self.staff = make_employee('staff1', is_staff=True)
self.regular = make_employee('user1')
self.target = make_employee('target1')
def test_anonymous_denied(self):
self.assertEqual(self.client.get('/api/employees/').status_code, 403)
def test_regular_user_denied(self):
self.client.force_login(self.regular)
self.assertEqual(self.client.get('/api/employees/').status_code, 403)
def test_staff_can_list(self):
self.client.force_login(self.staff)
resp = self.client.get('/api/employees/')
self.assertEqual(resp.status_code, 200)
def test_staff_cannot_create_via_api(self):
self.client.force_login(self.staff)
resp = self.client.post('/api/employees/', {
'username': 'hacker',
'first_name': 'H',
'last_name': 'H',
}, content_type='application/json')
self.assertEqual(resp.status_code, 405)
def test_staff_cannot_delete_via_api(self):
self.client.force_login(self.staff)
resp = self.client.delete(f'/api/employees/{self.target.pk}/')
self.assertEqual(resp.status_code, 405)
+10 -5
View File
@@ -1,12 +1,17 @@
from django.shortcuts import render
from rest_framework import viewsets
from rest_framework.permissions import IsAuthenticated
from rest_framework.permissions import IsAdminUser
from .models import Employee
from .serializers import EmployeeSerializer
# Create your views here.
class EmployeeViewSet(viewsets.ModelViewSet):
class EmployeeViewSet(viewsets.ReadOnlyModelViewSet):
"""
Read-only API сотрудников.
Полный CRUD через DRF закрыт: создание/изменение/удаление — только через
админку / HR-портал с проверками прав. Список и детали доступны staff.
"""
queryset = Employee.objects.all()
serializer_class = EmployeeSerializer
permission_classes = [IsAuthenticated]
permission_classes = [IsAdminUser]
@@ -57,10 +57,16 @@ def send_direct_message(employee, message: str, props: dict | None = None) -> bo
def _task_done_attachment(task) -> dict:
"""Интерактивна кнопка «Виконано» під повідомленням. task_id підписаний
(django.core.signing) — сервер перевіряє підпис у task_done_action,
окремого секрету заводити не треба, використовується SECRET_KEY."""
action_url = f'{settings.PORTAL_BASE_URL.rstrip("/")}/mattermost/actions/task-done/'
"""Интерактивна кнопка «Виконано» під повідомленням.
task_id підписаний (django.core.signing). URL містить shared secret
(MATTERMOST_INTERACTIVE_TOKEN) — task_done_action перевіряє обидва.
"""
from urllib.parse import urlencode
token = settings.MATTERMOST_INTERACTIVE_TOKEN
base = f'{settings.PORTAL_BASE_URL.rstrip("/")}/mattermost/actions/task-done/'
action_url = f'{base}?{urlencode({"token": token})}' if token else base
signed_task_id = signing.dumps(task.pk, salt=TASK_ACTION_SALT)
return {
'attachments': [{
+50
View File
@@ -256,3 +256,53 @@ class EnsurePersonalAccessTokenTests(TestCase):
self.assertEqual(token, 'already-stored-token')
service.create_user_access_token.assert_not_called()
class TaskDoneActionAuthTests(TestCase):
"""task_done_action требует MATTERMOST_INTERACTIVE_TOKEN + подпись."""
def setUp(self):
from django.core import signing
from apps.tasks.models import Task
from apps.mattermost_connector.services.bot_service import TASK_ACTION_SALT
self.employee = make_employee(mattermost_user_id='mm-user-1')
self.task = Task.objects.create(
title='Test task',
creator=self.employee,
status='in_progress',
)
self.task.responsible.add(self.employee)
self.signed = signing.dumps(self.task.pk, salt=TASK_ACTION_SALT)
def _post(self, *, token=None, query_token=None, user_id='mm-user-1'):
import json
url = '/mattermost/actions/task-done/'
if query_token is not None:
url = f'{url}?token={query_token}'
body = {
'user_id': user_id,
'context': {'t': self.signed},
}
if token is not None:
body['token'] = token
return self.client.post(
url, data=json.dumps(body), content_type='application/json',
)
def test_rejects_without_configured_token(self):
with self.settings(MATTERMOST_INTERACTIVE_TOKEN=''):
resp = self._post(query_token='x')
self.assertEqual(resp.status_code, 403)
def test_rejects_wrong_token(self):
with self.settings(MATTERMOST_INTERACTIVE_TOKEN='secret'):
resp = self._post(query_token='wrong')
self.assertEqual(resp.status_code, 403)
def test_accepts_query_token(self):
with self.settings(MATTERMOST_INTERACTIVE_TOKEN='secret'):
resp = self._post(query_token='secret')
self.assertEqual(resp.status_code, 200)
self.task.refresh_from_db()
self.assertEqual(self.task.status, 'done')
+19 -2
View File
@@ -57,13 +57,30 @@ def slash_command_tasks(request):
@csrf_exempt
@require_POST
def task_done_action(request):
"""Обработчик кнопки «Виконано» в интерактивному повідомленні бота.
https://developers.mattermost.com/integrate/interactive-messages/"""
"""Обработчик кнопки «Виконано» в інтерактивному повідомленні бота.
https://developers.mattermost.com/integrate/interactive-messages/
Fail-closed: потрібен MATTERMOST_INTERACTIVE_TOKEN (?token= у URL кнопки,
або поле token у JSON) + валідний підписаний task id.
"""
expected = settings.MATTERMOST_INTERACTIVE_TOKEN
if not expected:
return HttpResponseForbidden('invalid token')
try:
payload = json.loads(request.body)
except (json.JSONDecodeError, UnicodeDecodeError):
return HttpResponseForbidden('bad payload')
provided = (
request.GET.get('token')
or request.POST.get('token')
or payload.get('token')
or ''
)
if provided != expected:
return HttpResponseForbidden('invalid token')
from apps.mattermost_connector.services.bot_service import TASK_ACTION_SALT, TASK_ACTION_MAX_AGE
context = payload.get('context') or {}
+47 -8
View File
@@ -5,6 +5,7 @@ import base64
import io
from datetime import datetime
from django.db.models import Q
from django.http import HttpResponseRedirect, Http404
from PIL import Image
from rest_framework import serializers, status
@@ -13,6 +14,8 @@ from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from apps.employees.models import Employee
from ..services import register_attendance
from ..models import AttendanceLog
@@ -23,6 +26,26 @@ _ALLOWED_MIME_SIGNATURES = {
_MAX_PHOTO_BYTES = 10 * 1024 * 1024 # 10 MB
def _is_attendance_admin(user) -> bool:
return bool(user.is_staff or user.is_superuser)
def _can_access_employee_attendance(user, employee_id: int) -> bool:
"""Staff — любой; иначе только себя или прямого подчинённого."""
if _is_attendance_admin(user):
return True
if user.pk == employee_id:
return True
return Employee.objects.filter(pk=employee_id, manager_id=user.pk).exists()
def _scoped_attendance_qs(user):
qs = AttendanceLog.objects.select_related('employee', 'location')
if _is_attendance_admin(user):
return qs
return qs.filter(Q(employee_id=user.pk) | Q(employee__manager_id=user.pk))
class AttendanceCreateSerializer(serializers.Serializer):
employee_id = serializers.IntegerField(min_value=1)
location_id = serializers.IntegerField(min_value=1)
@@ -41,7 +64,6 @@ class AttendanceCreateSerializer(serializers.Serializer):
if len(photo_bytes) > _MAX_PHOTO_BYTES:
raise serializers.ValidationError('Фото превышает допустимый размер (10 МБ)')
# Проверяем сигнатуру файла (magic bytes)
mime = None
for sig, mime_type in _ALLOWED_MIME_SIGNATURES.items():
if photo_bytes[:len(sig)] == sig:
@@ -50,7 +72,6 @@ class AttendanceCreateSerializer(serializers.Serializer):
if mime is None:
raise serializers.ValidationError('Допускаются только JPEG и PNG изображения')
# Проверяем что файл действительно является изображением
try:
img = Image.open(io.BytesIO(photo_bytes))
img.verify()
@@ -78,6 +99,9 @@ class AttendanceRegisterView(APIView):
"""
POST /api/attendance/register/
Принимает фото в base64, создаёт запись посещения.
Обычный сотрудник — только за себя. Staff/superuser (учётки терминалов) —
за любого employee_id.
"""
authentication_classes = [SessionAuthentication]
permission_classes = [IsAuthenticated]
@@ -86,9 +110,16 @@ class AttendanceRegisterView(APIView):
serializer = AttendanceCreateSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
data = serializer.validated_data
employee_id = data['employee_id']
if not _is_attendance_admin(request.user) and request.user.pk != employee_id:
return Response(
{'error': 'Можна реєструвати відвідування лише за себе'},
status=status.HTTP_403_FORBIDDEN,
)
log = register_attendance(
employee_id=data['employee_id'],
employee_id=employee_id,
location_id=data['location_id'],
direction=data['direction'],
photo_bytes=data['photo_b64'],
@@ -102,18 +133,27 @@ class AttendanceListView(APIView):
"""
GET /api/attendance/?employee_id=&date=YYYY-MM-DD&location_id=
Список записей с превью (без оригиналов фото).
Staff — полный список (с фильтрами). Остальные — только свои логи
и логи прямых подчинённых.
"""
authentication_classes = [SessionAuthentication]
permission_classes = [IsAuthenticated]
def get(self, request):
qs = AttendanceLog.objects.select_related('employee', 'location')
qs = _scoped_attendance_qs(request.user)
if emp := request.query_params.get('employee_id'):
try:
qs = qs.filter(employee_id=int(emp))
employee_id = int(emp)
except (ValueError, TypeError):
return Response({'error': 'employee_id должен быть числом'}, status=status.HTTP_400_BAD_REQUEST)
if not _can_access_employee_attendance(request.user, employee_id):
return Response(
{'error': 'Немає доступу до відвідувань цього співробітника'},
status=status.HTTP_403_FORBIDDEN,
)
qs = qs.filter(employee_id=employee_id)
if loc := request.query_params.get('location_id'):
try:
@@ -135,7 +175,7 @@ class AttendanceListView(APIView):
class AttendancePhotoView(APIView):
"""
GET /api/attendance/<uuid>/photo/
Redirect к оригиналу фото — только для is_staff или руководителя сотрудника.
Redirect к оригиналу фото — staff, сам сотрудник или его руководитель.
"""
authentication_classes = [SessionAuthentication]
permission_classes = [IsAuthenticated]
@@ -149,8 +189,7 @@ class AttendancePhotoView(APIView):
if not log.photo:
raise Http404
user = request.user
if not (user.is_staff or user.pk == log.employee_id or user.pk == log.employee.manager_id):
if not _can_access_employee_attendance(request.user, log.employee_id):
raise Http404
return HttpResponseRedirect(log.photo.url)
+99
View File
@@ -166,3 +166,102 @@ class WorkingHoursTests(TestCase):
due = add_working_hours(start, 11)
self.assertEqual(due.date(), start.date() + timedelta(days=2))
self.assertEqual(due.hour, 11)
class AttendanceAPIAuthzTests(TestCase):
"""Object-level authz для /api/attendance/."""
@classmethod
def setUpTestData(cls):
from django.contrib.auth import get_user_model
from core.models import Directory, DirectoryType
from apps.locations.models import Pharmacy
import io
from PIL import Image
Employee = get_user_model()
org_type = DirectoryType.objects.create(code='organization', name='Организация')
region_type = DirectoryType.objects.create(code='region', name='Регион')
region = Directory.objects.create(directory_type=region_type, code='od', name='Одесса')
org = Directory.objects.create(directory_type=org_type, code='akc', name='АКС')
cls.location = Pharmacy.objects.create(
organization=org, location_region=region,
pharmacy_number=101, location_type='pharmacy',
)
cls.staff = Employee.objects.create_user(
username='att_staff', hire_date=date(2020, 1, 1),
first_name='S', last_name='Staff', is_staff=True,
mattermost_sync_enabled=False, ad_sync_enabled=False,
)
cls.alice = Employee.objects.create_user(
username='att_alice', hire_date=date(2020, 1, 1),
first_name='A', last_name='Alice',
mattermost_sync_enabled=False, ad_sync_enabled=False,
)
cls.bob = Employee.objects.create_user(
username='att_bob', hire_date=date(2020, 1, 1),
first_name='B', last_name='Bob', manager=cls.alice,
mattermost_sync_enabled=False, ad_sync_enabled=False,
)
cls.carol = Employee.objects.create_user(
username='att_carol', hire_date=date(2020, 1, 1),
first_name='C', last_name='Carol',
mattermost_sync_enabled=False, ad_sync_enabled=False,
)
buf = io.BytesIO()
Image.new('RGB', (8, 8), color='red').save(buf, format='JPEG')
cls.photo_b64 = __import__('base64').b64encode(buf.getvalue()).decode()
def _register(self, user, employee_id):
self.client.force_login(user)
return self.client.post(
'/api/attendance/register/',
{
'employee_id': employee_id,
'location_id': self.location.pk,
'direction': 'in',
'photo_b64': self.photo_b64,
},
content_type='application/json',
)
def test_user_can_register_self(self):
resp = self._register(self.alice, self.alice.pk)
self.assertEqual(resp.status_code, 201)
def test_user_cannot_register_other(self):
resp = self._register(self.alice, self.carol.pk)
self.assertEqual(resp.status_code, 403)
def test_staff_can_register_other(self):
resp = self._register(self.staff, self.carol.pk)
self.assertEqual(resp.status_code, 201)
def test_list_scoped_to_self_and_reports(self):
from apps.visit_manager.services import register_attendance
import base64
photo = base64.b64decode(self.photo_b64)
register_attendance(
employee_id=self.alice.pk, location_id=self.location.pk,
direction='in', photo_bytes=photo,
)
register_attendance(
employee_id=self.bob.pk, location_id=self.location.pk,
direction='in', photo_bytes=photo,
)
register_attendance(
employee_id=self.carol.pk, location_id=self.location.pk,
direction='in', photo_bytes=photo,
)
self.client.force_login(self.alice)
resp = self.client.get('/api/attendance/')
self.assertEqual(resp.status_code, 200)
ids = {row['employee_id'] for row in resp.json()}
self.assertIn(self.alice.pk, ids)
self.assertIn(self.bob.pk, ids)
self.assertNotIn(self.carol.pk, ids)
def test_list_other_employee_forbidden(self):
self.client.force_login(self.alice)
resp = self.client.get(f'/api/attendance/?employee_id={self.carol.pk}')
self.assertEqual(resp.status_code, 403)
+14 -1
View File
@@ -1,12 +1,25 @@
import base64
import hashlib
import logging
from django.conf import settings
from cryptography.fernet import Fernet
logger = logging.getLogger(__name__)
_warned_fallback = False
def _get_fernet() -> Fernet:
key_source = getattr(settings, 'FIELD_ENCRYPTION_KEY', None) or settings.SECRET_KEY
global _warned_fallback
key_source = getattr(settings, 'FIELD_ENCRYPTION_KEY', None) or ''
if not key_source:
key_source = settings.SECRET_KEY
if not _warned_fallback:
logger.warning(
'FIELD_ENCRYPTION_KEY is not set — falling back to SECRET_KEY. '
'Set a dedicated key in .env (production requires it).'
)
_warned_fallback = True
derived_key = base64.urlsafe_b64encode(hashlib.sha256(key_source.encode()).digest())
return Fernet(derived_key)
+2 -3
View File
@@ -12,7 +12,6 @@ from core.models.directories import Directory
from core.models import ConnectorConfig
from core.utils.htmx import render_page
from core.utils.pagination import paginate
import json
# Create your views here.
@@ -200,7 +199,7 @@ def pharmacies_view(request):
'pharmacies': pharmacies,
'active_count': active_count,
'regions': regions,
'region_tree_json': json.dumps(region_tree, ensure_ascii=False),
'region_tree': region_tree,
}
return render_page(request, 'pharmacies.html', context)
@@ -262,7 +261,7 @@ def dislocation_view(request):
'pharmacies_no_head': sum(1 for p in pharmacies if not p.head_pharmacist_id),
'pharmacies_empty': sum(1 for p in pharmacies if p.staff_count == 0),
'pharmacies_no_location': pharmacies_no_location,
'region_tree_json': json.dumps(region_tree, ensure_ascii=False),
'region_tree': region_tree,
'office_departments': office_departments,
'office_no_department': office_no_department,
'office_employees_count': len(office_employees),
+9 -9
View File
@@ -259,7 +259,14 @@
</svg>
<span class="nav-label-text">{% trans "Задачи" %}</span>
</a>
<a href="/dislocation/" class="nav-item"
hx-get="/dislocation/" hx-target="#main-content" hx-push-url="/dislocation/"
hx-indicator="#loader" onclick="setActive(this)">
<svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2">
<path d="M20 10c0 6-8 12-8 12s-8-6-8-12a8 8 0 0 1 16 0Z"/><circle cx="12" cy="10" r="3"/>
</svg>
<span class="nav-label-text">{% trans "Дислокация" %}</span>
</a>
<a href="/documents/" class="nav-item"
hx-get="/documents/" hx-target="#main-content" hx-push-url="/documents/"
hx-indicator="#loader" onclick="setActive(this)">
@@ -353,14 +360,7 @@
<div class="nav-section-label">{% trans "Управление" %}</div>
<div style="display:flex; flex-direction:column; gap:2px;">
<a href="/dislocation/" class="nav-item"
hx-get="/dislocation/" hx-target="#main-content" hx-push-url="/dislocation/"
hx-indicator="#loader" onclick="setActive(this)">
<svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2">
<path d="M20 10c0 6-8 12-8 12s-8-6-8-12a8 8 0 0 1 16 0Z"/><circle cx="12" cy="10" r="3"/>
</svg>
<span class="nav-label-text">{% trans "Дислокация" %}</span>
</a>
<a href="/structure/" class="nav-item"
hx-get="/structure/" hx-target="#main-content" hx-push-url="/structure/"
+1 -1
View File
@@ -342,7 +342,7 @@
</div>
<!-- ══ /ВКЛАДКА: ОФІС / СКЛАД ══ -->
<script id="dislo-region-tree-data" type="application/json">{{ region_tree_json|safe }}</script>
{{ region_tree|json_script:"dislo-region-tree-data" }}
<script src="https://cdn.jsdelivr.net/npm/leaflet@1.9.4/dist/leaflet.js"></script>
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/leaflet@1.9.4/dist/leaflet.css">
<script src="https://cdn.jsdelivr.net/npm/leaflet.markercluster@1.5.3/dist/leaflet.markercluster.js"></script>
+1 -1
View File
@@ -221,7 +221,7 @@
</div>
</div>
<script id="region-tree-data" type="application/json">{{ region_tree_json|safe }}</script>
{{ region_tree|json_script:"region-tree-data" }}
<script src="https://cdn.jsdelivr.net/npm/leaflet@1.9.4/dist/leaflet.js"></script>
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/leaflet@1.9.4/dist/leaflet.css">