diff --git a/src/edc_cdisc/constants.py b/src/edc_cdisc/constants.py index 891fa1a..13d5265 100644 --- a/src/edc_cdisc/constants.py +++ b/src/edc_cdisc/constants.py @@ -94,4 +94,27 @@ UNSCHEDULED_TYPE = "Unscheduled" COMMON_TYPE = "Common" +SCREENING_EVENT_CATEGORY = "screening" +CONSENT_EVENT_CATEGORY = "consent" + +# Models whose export field list is an explicit whitelist instead of the admin +# fieldsets. Used for documents that hold sensitive data (e.g. consent): only +# the named, non-PII fields are exported. The encrypted-field and +# EXCLUDED_FIELD_NAMES floors STILL apply on top (see iter_whitelist_fields), so +# an encrypted/excluded field added here cannot be re-exposed — it is dropped +# and a warning is emitted. Plaintext-sensitive additions are caught by review; +# keep these lists tiny and explicit. +CONSENT_EXPORT_FIELDS: tuple[str, ...] = ( + "subject_identifier", + "consent_datetime", + "model_name", + "version", + "consent_definition_name", +) + +# Static per-model whitelists (model label_lower -> field names). The consent +# model is resolved dynamically from settings.SUBJECT_CONSENT_MODEL in +# get_whitelist_fields, so it is not listed here. +WHITELIST_FIELDS: dict[str, tuple[str, ...]] = {} + ODM_SCHEMA_PATH = Path(__file__).parent / "odm_schema" / "cdisc-odm-1.3.1" / "ODM1-3-1.xsd" diff --git a/src/edc_cdisc/serializers/clinical_data_serializer.py b/src/edc_cdisc/serializers/clinical_data_serializer.py index 4f0981f..cedb5cb 100644 --- a/src/edc_cdisc/serializers/clinical_data_serializer.py +++ b/src/edc_cdisc/serializers/clinical_data_serializer.py @@ -103,6 +103,7 @@ def build_subject_data( if event_element is not None: event_elements.append(event_element) event_elements += self.build_common_event_data(subject_identifier) + event_elements += self.build_consent_event_data(subject_identifier) if not event_elements: return None @@ -187,6 +188,33 @@ def build_common_event_data(self, subject_identifier: str) -> list[etree._Elemen elements.append(element) return elements + def build_consent_event_data(self, subject_identifier: str) -> list[etree._Element]: + """Repeating Common event — one StudyEventData per consent row. + + Consent is unique on subject_identifier + version, so a subject may have + several rows; each is emitted with StudyEventRepeatKey = version. Only + the whitelist fields (CONSENT_EXPORT_FIELDS) are serialized — the consent + document holds sensitive data. + """ + consent_model = self.get_consent_model() + if not consent_model: + return [] + elements: list[etree._Element] = [] + qs = ( + django_apps.get_model(consent_model) + .objects.filter(subject_identifier=subject_identifier) + .order_by("version", "consent_datetime") + ) + for instance in qs: + element = etree.Element( + "StudyEventData", + StudyEventOID=self.common_event_oid(consent_model), + StudyEventRepeatKey=str(instance.version), + ) + element.append(self.build_form_data(consent_model, instance)) + elements.append(element) + return elements + @staticmethod def get_crfs_for_visit( subject_visit: RelatedVisitProtocol, diff --git a/src/edc_cdisc/serializers/metadata_serializer.py b/src/edc_cdisc/serializers/metadata_serializer.py index 35455fa..e0daa61 100644 --- a/src/edc_cdisc/serializers/metadata_serializer.py +++ b/src/edc_cdisc/serializers/metadata_serializer.py @@ -10,10 +10,12 @@ from ..constants import ( CODELIST, COMMON_TYPE, + CONSENT_EVENT_CATEGORY, FORM, ITEM, ITEM_GROUP, SCHEDULED_TYPE, + SCREENING_EVENT_CATEGORY, UNSCHEDULED_TYPE, ) from ..exceptions import ProtocolSerializerError @@ -70,14 +72,14 @@ def build_metadata_version(self) -> etree._Element: element = etree.Element("MetaDataVersion") element.append(self.get_protocol_element()) - screening_model = self.get_screening_model() + subject_level_events = list(self.iter_subject_level_events()) # StudyEventDefs (order among them is free; schedule carried by Alias) - for model in self.get_common_models(): - element.append(self.get_common_event_def_element(model)) - if screening_model: + for model, category, repeating in subject_level_events: element.append( - self.get_common_event_def_element(screening_model, event_category="screening") + self.get_common_event_def_element( + model, event_category=category, repeating=repeating + ) ) for schedule in self.visit_schedule.schedules.values(): for visit in schedule.visits.values(): @@ -95,9 +97,7 @@ def build_metadata_version(self) -> etree._Element: for visit in schedule.visits.values() for crf in [*visit.all_crfs, *visit.all_requisitions] } - common_models = [*self.get_common_models()] - if screening_model: - common_models.append(screening_model) + common_models = [model for model, _category, _repeating in subject_level_events] models = [ m for m in dict.fromkeys([*common_models, *sorted(form_models)]) @@ -130,22 +130,30 @@ def get_protocol_element(self) -> etree._Element: StudyEventOID=self.unscheduled_event_oid(visit_code), Mandatory=NO, ) - for model in self.get_common_models(): + for model, _category, _repeating in self.iter_subject_level_events(): etree.SubElement( element, "StudyEventRef", StudyEventOID=self.common_event_oid(model), Mandatory=NO, ) - if screening_model := self.get_screening_model(): - etree.SubElement( - element, - "StudyEventRef", - StudyEventOID=self.common_event_oid(screening_model), - Mandatory=NO, - ) return element + def iter_subject_level_events(self) -> Iterator[tuple[str, str | None, bool]]: + """Yield ``(model, event_category, repeating)`` for each non-scheduled, + subject-level Common event: death/offstudy, then screening, then consent. + + Single source for the StudyEventDefs, the definition catalog, and the + Protocol StudyEventRefs so they cannot drift. Consent is repeating + (unique on subject_identifier + version); the rest are singletons. + """ + for model in self.get_common_models(): + yield model, None, False + if screening_model := self.get_screening_model(): + yield screening_model, SCREENING_EVENT_CATEGORY, False + if consent_model := self.get_consent_model(): + yield consent_model, CONSENT_EVENT_CATEGORY, True + @staticmethod def append_schedule_alias(element: etree._Element, schedule_name: str) -> None: etree.SubElement(element, "Alias", Context="clinicedc.schedule", Name=schedule_name) @@ -326,13 +334,14 @@ def get_common_event_def_element( self, model: str, event_category: str | None = None, + repeating: bool = False, ) -> etree._Element: verbose_name = str(django_apps.get_model(model)._meta.verbose_name) element = etree.Element( "StudyEventDef", OID=self.common_event_oid(model), Name=verbose_name, - Repeating=NO, + Repeating=YES if repeating else NO, Type=COMMON_TYPE, ) etree.SubElement( diff --git a/src/edc_cdisc/serializers/visit_schedule_serializer.py b/src/edc_cdisc/serializers/visit_schedule_serializer.py index 5f5669d..1b5286e 100644 --- a/src/edc_cdisc/serializers/visit_schedule_serializer.py +++ b/src/edc_cdisc/serializers/visit_schedule_serializer.py @@ -91,6 +91,17 @@ def get_screening_model(self) -> str | None: model = getattr(settings, "SUBJECT_SCREENING_MODEL", None) return model if model and self._model_exists(model) else None + def get_consent_model(self) -> str | None: + """The consent model (settings.SUBJECT_CONSENT_MODEL). + + A *repeating* Common event keyed by subject_identifier (the allocation + identifier consent assigns), with repeat key = ``version`` — one row per + consent version. The consent document holds sensitive data, so only the + explicit whitelist (CONSENT_EXPORT_FIELDS) is exported. + """ + model = getattr(settings, "SUBJECT_CONSENT_MODEL", None) + return model if model and self._model_exists(model) else None + @staticmethod def _model_exists(model: str) -> bool: try: diff --git a/src/edc_cdisc/tests/tests/test_validate_odm.py b/src/edc_cdisc/tests/tests/test_validate_odm.py index 197871e..8e757c4 100644 --- a/src/edc_cdisc/tests/tests/test_validate_odm.py +++ b/src/edc_cdisc/tests/tests/test_validate_odm.py @@ -4,7 +4,7 @@ import time_machine from clinicedc_tests.consents import consent_v1 from clinicedc_tests.helper import Helper -from clinicedc_tests.models import CrfFour, SubjectScreening +from clinicedc_tests.models import CrfFour, SubjectConsent, SubjectScreening from dateutil.relativedelta import relativedelta from django.contrib import admin from django.db import models @@ -28,7 +28,9 @@ MetadataSerializer, SnapshotSerializer, ) -from edc_cdisc.utils import is_encrypted_field, validate_odm +from edc_cdisc.utils import is_encrypted_field, iter_whitelist_fields, validate_odm + +CONSENT_OID = "CE.clinicedc_tests.subjectconsent" NS = {"odm": ODM_NAMESPACE} @@ -176,6 +178,56 @@ def test_screening_event_data_present(self) -> None: self.assertIsNotNone(sed) self.assertEqual(validate_odm(self._build(SnapshotSerializer)), []) + def test_consent_event_def_is_repeating_with_category_alias(self) -> None: + root = etree.fromstring(self._build(MetadataSerializer)) + sed = root.find(f".//odm:StudyEventDef[@OID='{CONSENT_OID}']", NS) + self.assertIsNotNone(sed) + self.assertEqual(sed.get("Type"), "Common") + # consent is unique on subject_identifier + version → repeating + self.assertEqual(sed.get("Repeating"), "Yes") + alias = sed.find("odm:Alias[@Context='clinicedc.event_category']", NS) + self.assertIsNotNone(alias) + self.assertEqual(alias.get("Name"), "consent") + + def test_consent_event_data_present_with_repeat_key(self) -> None: + # enroll_to_baseline created a consent for the enrolled subject + root = etree.fromstring(self._build(SnapshotSerializer)) + sed = root.find( + f".//odm:ClinicalData//odm:StudyEventData[@StudyEventOID='{CONSENT_OID}']", NS + ) + self.assertIsNotNone(sed) + # repeat key carries the consent version + self.assertTrue(sed.get("StudyEventRepeatKey")) + self.assertEqual(validate_odm(self._build(SnapshotSerializer)), []) + + def test_consent_export_drops_encrypted_pii(self) -> None: + # only the whitelist is exported; encrypted PII (e.g. first_name) must + # never appear as an ItemDef or ItemData + meta = etree.fromstring(self._build(MetadataSerializer)) + data = etree.fromstring(self._build(SnapshotSerializer)) + for root in (meta, data): + oids = { + e.get("ItemOID") or e.get("OID") + for e in root.iter() + if (e.get("ItemOID") or e.get("OID")) + } + self.assertNotIn("I.clinicedc_tests.subjectconsent.first_name", oids) + self.assertIn("I.clinicedc_tests.subjectconsent.version", oids) + + +class TestConsentWhitelistFloor(TestCase): + def test_floor_drops_encrypted_field_even_if_whitelisted(self) -> None: + # first_name is an encrypted PII field; whitelisting it must not export + # it — it is dropped (with a warning) while real fields pass through + with self.assertWarns(UserWarning): + fields = list( + iter_whitelist_fields( + SubjectConsent, ["subject_identifier", "first_name", "version"] + ) + ) + names = {f.name for f in fields} + self.assertEqual(names, {"subject_identifier", "version"}) + class TestEncryptedFieldDetection(TestCase): def test_is_encrypted_field(self) -> None: diff --git a/src/edc_cdisc/utils.py b/src/edc_cdisc/utils.py index 65b1f60..a1255b7 100644 --- a/src/edc_cdisc/utils.py +++ b/src/edc_cdisc/utils.py @@ -5,6 +5,7 @@ from datetime import date, datetime, time from typing import TYPE_CHECKING +from django.conf import settings from django.contrib import admin from django.contrib.admin.sites import all_sites from django.db import models @@ -12,10 +13,12 @@ from lxml import etree from .constants import ( + CONSENT_EXPORT_FIELDS, DJANGO_TO_ODM_DATATYPE, EXCLUDED_FIELD_NAMES, EXCLUDED_FIELDSET_NAMES, ODM_SCHEMA_PATH, + WHITELIST_FIELDS, ) from .exceptions import ModelAdminNotFoundError @@ -129,6 +132,44 @@ def is_encrypted_field(field: models.Field) -> bool: return isinstance(field, BaseField) +def get_whitelist_fields(model_label: str) -> tuple[str, ...] | None: + """Explicit export field list for ``model_label``, or ``None`` when the + model is driven by its admin fieldsets. + + The consent model (``settings.SUBJECT_CONSENT_MODEL``) holds sensitive data, + so only the explicit, non-PII :data:`CONSENT_EXPORT_FIELDS` are exported. + Static whitelists for other models live in :data:`WHITELIST_FIELDS`. + """ + if model_label == getattr(settings, "SUBJECT_CONSENT_MODEL", None): + return CONSENT_EXPORT_FIELDS + return WHITELIST_FIELDS.get(model_label) + + +def iter_whitelist_fields( + model_cls: type[models.Model], field_names: Iterable[str] +) -> Iterator[models.Field]: + """Yield the explicitly whitelisted fields of ``model_cls``. + + The same PII floor as everywhere else still applies: an encrypted field or + an :data:`EXCLUDED_FIELD_NAMES` field is dropped *even if whitelisted*, with + a warning so the mistake is visible. This is what stops a sensitive field + from being re-exposed by being added to a whitelist. A name that is not a + real field raises ``FieldDoesNotExist`` (whitelists must be explicit). + """ + label = model_cls._meta.label_lower + for field_name in field_names: + field = model_cls._meta.get_field(field_name) + if is_encrypted_field(field) or field.name in EXCLUDED_FIELD_NAMES: + warnings.warn( + f"Whitelisted field {label}.{field_name} is encrypted or " + "excluded; not exported.", + UserWarning, + stacklevel=2, + ) + continue + yield field + + def iter_crf_sections( model_cls: type[models.Model], ) -> Iterator[tuple[int, str | None, list[models.Field]]]: @@ -141,9 +182,20 @@ def iter_crf_sections( positional index and is kept stable (it is not renumbered when a section drops out) so ``ItemGroupOID``\\s do not shift. + Whitelisted models (see :func:`get_whitelist_fields`) bypass the admin + entirely and yield a single unnamed section of their explicit fields — so a + model holding sensitive data needs no admin registered and exports only the + vetted list. + Note: relation fields (FK / O2O / M2M) are *included* for now (emitted as text); proper relation handling is a later step. """ + whitelist = get_whitelist_fields(model_cls._meta.label_lower) + if whitelist is not None: + fields = list(iter_whitelist_fields(model_cls, whitelist)) + if fields: + yield 1, None, fields + return for index, (name, opts) in enumerate(get_modeladmin_fieldsets(model_cls), start=1): fields = [] for field_name in opts.get("fields"):