ixgadmin
8ad83a7a65
Batch writes were closed in 88eede5, but every other app's viewset still used
IsAuthenticated, which in this platform means any buyer. A buyer could rewrite a
document record, create a QC result (which moves the batch status and raises a
CAPA), and reach the write endpoints for CAPAs, certificates, ESG reports and
shipments.
- Use the one IsOpsOrAdmin from apps.compliance.permissions; drop the duplicate
copy in apps/qc/views.py.
- Apply it to the whole viewset on shipments, qc, capa, esg, documents,
certificates-of-analysis and regulatory-certs. The buyer portal is
server-rendered through the ORM, so it needs none of these APIs.
- Keep the buyer chat working: AI interactions stay authenticated and scoped,
and agent reads stay open while agent writes become ops/admin.
- Close a buyer-to-admin escalation in the users API: writes are admin-only and
role/is_active are read-only, so a buyer can no longer PATCH their own row to
role=admin and then pass every IsOpsOrAdmin check.
105 lines
4.6 KiB
Python
105 lines
4.6 KiB
Python
import os
|
|
import uuid
|
|
from django.conf import settings
|
|
from rest_framework import viewsets, status
|
|
from rest_framework.decorators import action
|
|
from rest_framework.parsers import MultiPartParser, FormParser
|
|
from rest_framework.response import Response
|
|
|
|
from apps.compliance.permissions import IsOpsOrAdmin
|
|
|
|
from .models import Document, CertificateOfAnalysis, RegulatoryCert
|
|
from .serializers import DocumentSerializer, DocumentUploadSerializer, CertificateOfAnalysisSerializer, RegulatoryCertSerializer
|
|
|
|
|
|
class DocumentViewSet(viewsets.ModelViewSet):
|
|
queryset = Document.objects.all()
|
|
serializer_class = DocumentSerializer
|
|
# Documents and certificates are the evidence a buyer's auditor relies on.
|
|
# The buyer portal serves the buyer-visible rows through the ORM
|
|
# (apps/buyer/views.py), so this API needs no buyer access — read or write.
|
|
permission_classes = [IsOpsOrAdmin]
|
|
|
|
@action(detail=False, methods=["post"], parser_classes=[MultiPartParser, FormParser])
|
|
def upload(self, request):
|
|
serializer = DocumentUploadSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
|
|
file = serializer.validated_data["file"]
|
|
batch_id = serializer.validated_data.get("batch_id")
|
|
doc_type = serializer.validated_data["doc_type"]
|
|
visibility = serializer.validated_data.get("visibility", "buyer")
|
|
|
|
file_ext = file.name.split(".")[-1].lower()
|
|
allowed = ["pdf", "jpg", "jpeg", "png", "xlsx"]
|
|
if file_ext not in allowed:
|
|
return Response({"error": f"File type .{file_ext} not allowed"}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
if file.size > 50 * 1024 * 1024:
|
|
return Response({"error": "File exceeds 50MB limit"}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
batch_dir = os.path.join(settings.MEDIA_ROOT, "ixg-documents", str(batch_id or "uncategorized"))
|
|
os.makedirs(batch_dir, exist_ok=True)
|
|
|
|
filename = f"{uuid.uuid4()}_{file.name}"
|
|
filepath = os.path.join(batch_dir, filename)
|
|
with open(filepath, "wb+") as dest:
|
|
for chunk in file.chunks():
|
|
dest.write(chunk)
|
|
|
|
doc = Document.objects.create(
|
|
batch_id=batch_id,
|
|
doc_type=doc_type,
|
|
file_url=f"/media/ixg-documents/{batch_id or 'uncategorized'}/{filename}",
|
|
file_name=file.name,
|
|
file_size_bytes=file.size,
|
|
visibility=visibility,
|
|
uploaded_by=request.user,
|
|
)
|
|
|
|
return Response(DocumentSerializer(doc).data, status=status.HTTP_201_CREATED)
|
|
|
|
|
|
class CertificateOfAnalysisViewSet(viewsets.ModelViewSet):
|
|
queryset = CertificateOfAnalysis.objects.select_related("batch", "qc_result").all()
|
|
serializer_class = CertificateOfAnalysisSerializer
|
|
# A certificate is released to buyers through the ORM (apps/buyer/views.py),
|
|
# not this API, so the API needs no buyer access. Only a published
|
|
# certificate is buyer-visible at all, and only ops/admin may publish one.
|
|
permission_classes = [IsOpsOrAdmin]
|
|
|
|
@action(detail=True, methods=["post"])
|
|
def approve(self, request, pk=None):
|
|
"""Commercial release approval: publishes the CoA and exposes it to buyers."""
|
|
from apps.compliance.services import publish_coa
|
|
|
|
coa = self.get_object()
|
|
if coa.status != CertificateOfAnalysis.Status.PENDING_APPROVAL:
|
|
return Response({"detail": "CoA is not pending approval"}, status=status.HTTP_400_BAD_REQUEST)
|
|
publish_coa(coa, released_by=request.user)
|
|
return Response(CertificateOfAnalysisSerializer(coa).data)
|
|
|
|
@action(detail=True, methods=["post"])
|
|
def supersede(self, request, pk=None):
|
|
from django.utils import timezone
|
|
|
|
coa = self.get_object()
|
|
replacement_id = request.data.get("replacement_id")
|
|
if not replacement_id:
|
|
return Response({"detail": "replacement_id is required"}, status=status.HTTP_400_BAD_REQUEST)
|
|
replacement = CertificateOfAnalysis.objects.filter(id=replacement_id).first()
|
|
if not replacement:
|
|
return Response({"detail": "Replacement CoA not found"}, status=status.HTTP_400_BAD_REQUEST)
|
|
coa.status = CertificateOfAnalysis.Status.SUPERSEDED
|
|
coa.superseded_at = timezone.now()
|
|
coa.superseded_by = replacement
|
|
coa.save()
|
|
return Response(CertificateOfAnalysisSerializer(coa).data)
|
|
|
|
|
|
class RegulatoryCertViewSet(viewsets.ModelViewSet):
|
|
queryset = RegulatoryCert.objects.all()
|
|
serializer_class = RegulatoryCertSerializer
|
|
# Regulatory certificates are an internal compliance record; the buyer portal
|
|
# does not read them, so the API is ops/admin only.
|
|
permission_classes = [IsOpsOrAdmin]
|