AB#207787 Information Exposure

backend/hct_mis_api/apps/household/views.py

@@ -1,4 +1,5 @@
-from typing import Dict, Optional
+import logging
+from typing import Dict, Optional, Union
 
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.request import Request
@@ -12,19 +13,21 @@
     IDENTIFICATION_TYPE_TAX_ID,
     Document,
     Household,
+    Individual,
+    PendingDocument,
+    PendingIndividual,
 )
 from hct_mis_api.apps.household.serializers import (
     serialize_by_household,
     serialize_by_individual,
 )
-from hct_mis_api.apps.registration_datahub.models import (
-    ImportedDocument,
-    ImportedHousehold,
-)
+from hct_mis_api.apps.registration_datahub.models import PendingHousehold
 from hct_mis_api.apps.utils.profiling import profiling
 
+logger = logging.getLogger(__name__)
+
 
-def get_individual(tax_id: str, business_area_code: Optional[str]) -> Document:
+def get_individual(tax_id: str, business_area_code: Optional[str]) -> Union[Individual, PendingIndividual]:
     documents = (
         Document.objects.all()
         if not business_area_code
@@ -35,21 +38,21 @@ def get_individual(tax_id: str, business_area_code: Optional[str]) -> Document:
     if documents.count() == 1:
         return documents.first().individual
 
-    imported_documents = (
-        ImportedDocument.objects.all()
+    pending_documents = (
+        PendingDocument.objects.all()
         if not business_area_code
-        else ImportedDocument.objects.filter(
+        else PendingDocument.objects.filter(
             individual__household__registration_data_import__business_area__code=business_area_code
         )
     ).filter(type__key=IDENTIFICATION_TYPE_TO_KEY_MAPPING[IDENTIFICATION_TYPE_TAX_ID], document_number=tax_id)
-    if imported_documents.count() > 1:
-        raise Exception(f"Multiple imported documents ({imported_documents.count()}) with given tax_id found")
-    if imported_documents.count() == 1:
-        return imported_documents.first().individual
+    if pending_documents.count() > 1:
+        raise Exception(f"Multiple imported documents ({pending_documents.count()}) with given tax_id found")
+    if pending_documents.count() == 1:
+        return pending_documents.first().individual
     raise Exception("Document with given tax_id not found")
 
 
-def get_household(registration_id: str, business_area_code: Optional[str]) -> ImportedHousehold:
+def get_household(registration_id: str, business_area_code: Optional[str]) -> Union[PendingHousehold, Household]:
     kobo_asset_value = _prepare_kobo_asset_id_value(registration_id)
     households = (
         Household.objects.all()
@@ -62,17 +65,17 @@ def get_household(registration_id: str, business_area_code: Optional[str]) -> Im
         return households.first()  # type: ignore
 
     if business_area_code is None:
-        imported_households_by_business_area = ImportedHousehold.objects.all()
+        pending_households_by_business_area = PendingHousehold.objects.all()
     else:
         business_areas = BusinessArea.objects.filter(code=business_area_code)
         if not business_areas:
             raise Exception(f"Business area with code {business_area_code} not found")
         business_area = business_areas.first()  # code is unique, so no need to worry here
-        imported_households_by_business_area = ImportedHousehold.objects.filter(
+        pending_households_by_business_area = PendingHousehold.objects.filter(
             registration_data_import__business_area_slug=business_area.slug
         )
 
-    imported_households = imported_households_by_business_area.filter(detail_id__endswith=kobo_asset_value)
+    imported_households = pending_households_by_business_area.filter(detail_id__endswith=kobo_asset_value)
     if imported_households.count() > 1:
         raise Exception(
             f"Multiple imported households ({imported_households.count()}) with given registration_id found"
@@ -112,7 +115,8 @@ def get(self, request: Request) -> Response:
 
         try:
             data = get_household_or_individual(tax_id, registration_id, business_area_code)
-        except Exception as exception:
-            return Response({"status": "not found", "error_message": str(exception)}, status=404)
+        except Exception as e:  # pragma: no cover
+            logger.exception(e)
+            return Response({"status": "not found", "error_message": "Household not Found"}, status=404)
 
         return Response(data, status=200)

backend/hct_mis_api/apps/registration_datahub/admin.py

@@ -24,8 +24,8 @@
     RegistrationDataImportDatahub,
 )
 from hct_mis_api.apps.registration_datahub.models import (
-    ImportedHousehold,
     ImportedIndividual,
+    PendingHousehold,
 )
 from hct_mis_api.apps.utils.admin import HOPEModelAdminBase
 
@@ -67,7 +67,7 @@ def inspect(self, request: HttpRequest, pk: UUID) -> TemplateResponse:
         context["title"] = f"Import {obj.name} - {obj.import_done}"
         context["data"] = {}
         has_content = False
-        for model in [ImportedIndividual, ImportedHousehold]:
+        for model in [ImportedIndividual, PendingHousehold]:
             count = model.objects.filter(registration_data_import=obj).count()
             has_content = has_content or count
             context["data"][model] = {"count": count, "warnings": [], "errors": [], "meta": model._meta}

backend/hct_mis_api/apps/registration_datahub/fixtures.py

@@ -22,9 +22,9 @@
     ImportedBankAccountInfo,
     ImportedDocument,
     ImportedDocumentType,
-    ImportedHousehold,
     ImportedIndividual,
     ImportedIndividualIdentity,
+    PendingHousehold,
     Record,
     RegistrationDataImportDatahub,
 )
@@ -48,7 +48,7 @@ class Meta:
 
 class ImportedHouseholdFactory(DjangoModelFactory):
     class Meta:
-        model = ImportedHousehold
+        model = PendingHousehold
 
     consent_sign = factory.django.ImageField(color="blue")
     consent = True
@@ -138,7 +138,7 @@ class Meta:
 
 def create_imported_household(
     household_args: Optional[Dict] = None, individual_args: Optional[Dict] = None
-) -> Tuple[ImportedHousehold, ImportedIndividual]:
+) -> Tuple[PendingHousehold, ImportedIndividual]:
     if household_args is None:
         household_args = {}
     if individual_args is None:
@@ -154,12 +154,12 @@ def create_imported_household(
 
 def create_imported_household_and_individuals(
     household_data: Optional[Dict] = None, individuals_data: Optional[List[Dict]] = None
-) -> Tuple[ImportedHousehold, List[ImportedIndividual]]:
+) -> Tuple[PendingHousehold, List[ImportedIndividual]]:
     if household_data is None:
         household_data = {}
     if individuals_data is None:
         individuals_data = []
-    household: ImportedHousehold = ImportedHouseholdFactory.build(**household_data, size=len(individuals_data))
+    household: PendingHousehold = ImportedHouseholdFactory.build(**household_data, size=len(individuals_data))
     individuals: List[ImportedIndividual] = [
         ImportedIndividualFactory(household=household, **individual_data) for individual_data in individuals_data
     ]

backend/hct_mis_api/apps/registration_datahub/models.py

@@ -83,7 +83,7 @@
 )
 
 
-class ImportedHousehold(TimeStampedUUIDModel):
+class PendingHousehold(TimeStampedUUIDModel):
     class CollectType(models.TextChoices):
         STANDARD = "STANDARD", "Standard"
         SINGLE = "SINGLE", "Single"
@@ -239,7 +239,7 @@ class ImportedIndividual(TimeStampedUUIDModel):
     email = models.CharField(max_length=255, blank=True)
     payment_delivery_phone_no = PhoneNumberField(blank=True, default=BLANK)
     household = models.ForeignKey(
-        "ImportedHousehold",
+        "hct_mis_api.apps.registration_datahub.models.PendingHousehold",
         null=True,
         related_name="individuals",
         on_delete=models.CASCADE,
@@ -353,7 +353,7 @@ class ImportedIndividualRoleInHousehold(TimeStampedUUIDModel):
         related_name="households_and_roles",
     )
     household = models.ForeignKey(
-        "ImportedHousehold",
+        "hct_mis_api.apps.registration_datahub.models.PendingHousehold",
         on_delete=models.CASCADE,
         related_name="individuals_and_roles",
     )
@@ -517,7 +517,7 @@ class KoboImportedSubmission(models.Model):
     kobo_submission_time = models.DateTimeField()  # ImportedHousehold.kobo_submission_time
     # we use on_delete=models.SET_NULL because we want to be able to delete
     # ImportedHousehold without loosing track of importing
-    imported_household = models.ForeignKey(ImportedHousehold, blank=True, null=True, on_delete=models.SET_NULL)
+    imported_household = models.ForeignKey(PendingHousehold, blank=True, null=True, on_delete=models.SET_NULL)
     amended = models.BooleanField(default=False, blank=True)
 
     registration_data_import = models.ForeignKey(