#773

Committed 30 Jul 2026 07:05AM UTC coverage: 88.786% (+0.3%) from 88.523%

Build # #773

Build Type

Pull #280

coveralls-python

Committed by

web-flow

Commit Message

[#271] Enforce tenant ownership on every write (#282)

* [#271] Add the tenant stamping mixin and scoped related field

Two write-side mechanisms mirroring the read-side for_user. The
stamping mixin sets a new row's tenant from the authenticated user
(never the payload) in create(). The scoped related field narrows a
foreign-key input's candidates to for_user(user), so a pk pointing at
another tenant's object fails validation as does_not_exist. Both
resolve the acting user from either the "user" or the DRF "request"
context key.

A test covers the case that matters most for trust: a payload carrying
an explicit tenant id is ignored, and the row is stamped to the caller.

* [#271] Stamp direct-FK creates with the creator's tenant

Form, organisation, user, entity, attribute and single-administration
creates now set tenant from the authenticated user. Rows that landed
tenant-less before were invisible to their own creator the moment read
filtering arrived — a form saved in the builder and vanished. A test
covers exactly that: create, then list, and find it.

The plan expected two variants, the mixin where create() is default and
explicit assignment where it is overridden. Only the mixin is needed:
all three custom create() methods call super().create(), so the mixin
sits behind them in the MRO and stamps on the way through. One
mechanism, applied uniformly.

Two unscoped level lookups had to be fixed to get here. validate_parent
and _assign_level both resolved the child tier with
Levels.objects.get(level=n), which is unique per tenant but not
globally: with a second tenant present they matched two rows and raised
MultipleObjectsReturned, so creating any child administration was a 500
rather than a leak. Both now resolve within the parent's tenant. Read
filtering missed them because they live in a serializer validator, not
a view queryset.

* [#271] Reject write payloads referencing another tenant's objects

Foreign-key inputs to tenant-owned objects — a... (continued)

Pull Request Pull Request #280: Epic/multi tenant saas

Coverage Stats

5733 of 6629 branches covered (86.48%)

Branch coverage included in aggregate %.

10877 of 12079 relevant lines covered (90.05%)

0.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.43

backend/utils/custom_serializer_fields.py

from django.core.validators import EmailValidator

from utils.tenant_scoped_model import acting_user
from django.utils.translation import gettext_lazy as _
from rest_framework.fields import (
    IntegerField,
    ChoiceField,
    CharField,
    ImageField,
    ListField,
    BooleanField,
    FloatField,
    DecimalField,
    URLField,
    DateField,
    MultipleChoiceField,
    FileField,
    DateTimeField,
    JSONField,
    Field,
)
from rest_framework.relations import PrimaryKeyRelatedField

key_map = {}


class CustomIntegerField(IntegerField):
    default_error_messages = {
        "invalid": _("A valid field_title is required."),
        "max_value": _(
            "Ensure field_title is less than or equal to {max_value}."
        ),
        "min_value": _(
            "Ensure field_title is greater than or equal to {min_value}."
        ),
        "max_string_length": _("field_title value too large."),
        "required": _("field_title is required."),
        "null": _("field_title may not be null."),
    }


class CustomCharField(CharField):
    default_error_messages = {
        "invalid": _("A valid field_title is required."),
        "blank": _("field_title may not be blank."),
        "max_length": _(
            "Ensure field_title has no more than {max_length} characters."
        ),
        "min_length": _(
            "Ensure field_title has at least {min_length} characters."
        ),
        "required": _("field_title is required."),
        "null": _("field_title may not be null."),
    }


class CustomChoiceField(ChoiceField):
    default_error_messages = {
        "invalid_choice": _('"{input}" is not a valid choice in field_title.'),
        "required": _("field_title is required."),
        "null": _("field_title may not be null."),
    }


class CustomMultipleChoiceField(MultipleChoiceField):
    default_error_messages = {
        "invalid_choice": _('"{input}" is not a valid choice in field_title.'),
        "not_a_list": _(
            'Expected a list of items but got type "{input_type}"'
            " in field_title."
        ),
        "empty": _("This selection may not be empty in field_title."),
    }


class CustomImageField(ImageField):
    default_error_messages = {
        "invalid_image": _(
            "Upload a valid image. field_title you uploaded was either not "
            "an image or a corrupted image."
        ),
        "required": _("field_title is required."),
        "null": _("field_title may not be null."),
    }


class CustomEmailField(CustomCharField):
    default_error_messages = {"invalid": _("Enter a valid email address.")}

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        validator = EmailValidator(message=self.error_messages["invalid"])
        self.validators.append(validator)


class CustomListField(ListField):
    default_error_messages = {
        "not_a_list": _(
            "Expected a list of items in field_title but got type"
            ' "{input_type}".'
        ),
        "empty": _("field_title may not be empty."),
        "min_length": _(
            "Ensure field_title has at least {min_length} elements."
        ),
        "max_length": _(
            "Ensure field_title has no more than {max_length} elements."
        ),
        "required": _("field_title is required."),
        "null": _("field_title may not be null."),
    }


class CustomBooleanField(BooleanField):
    default_error_messages = {
        "invalid": _("Must be a valid field_title."),
        "required": _("field_title is required."),
        "null": _("field_title may not be null."),
    }


class CustomFloatField(FloatField):
    default_error_messages = {
        "invalid": _("A valid field_title is required."),
        "max_value": _(
            "Ensure field_title is less than or equal to {max_value}."
        ),
        "min_value": _(
            "Ensure field_title is greater than or equal to {min_value}."
        ),
        "max_string_length": _("field_title too large."),
        "required": _("field_title is required."),
        "null": _("field_title may not be null."),
    }


class CustomDecimalField(DecimalField):
    default_error_messages = {
        "invalid": _("A field_title number is required."),
        "max_value": _(
            "Ensure field_title is less than or equal to {max_value}."
        ),
        "min_value": _(
            "Ensure field_title is greater than or equal to {min_value}."
        ),
        "max_digits": _(
            "Ensure that in field_title are no more than {max_digits} "
            "digits in total."
        ),
        "max_decimal_places": _(
            "Ensure that in field_title are no more than {max_decimal_places} "
            "decimal places."
        ),
        "max_whole_digits": _(
            "Ensure that in field_title are no more than {max_whole_digits} "
            "digits before the "
            "decimal point."
        ),
        "max_string_length": _("String value too large in field_title."),
        "required": _("field_title is required."),
        "null": _("field_title may not be null."),
    }


class CustomURLField(URLField):
    default_error_messages = {
        "invalid": _("Enter a valid URL in field_title."),
        "required": _("field_title is required."),
        "null": _("field_title may not be null."),
    }


class CustomPrimaryKeyRelatedField(PrimaryKeyRelatedField):
    default_error_messages = {
        "required": _("field_title is required."),
        "does_not_exist": _(
            'Invalid pk "{pk_value}" - object does not exist.'
        ),
        "incorrect_type": _(
            "Incorrect type. Expected pk value, received {data_type}."
        ),
        "null": _("field_title may not be null."),
    }


class CustomDateField(DateField):
    default_error_messages = {
        "required": _("field_title is required."),
        "invalid": _(
            "Date has wrong format. Use one of these formats instead:"
            " {format}."
        ),
        "datetime": _("Expected a date but got a datetime."),
        "null": _("field_title may not be null."),
    }


class CustomFileField(FileField):
    default_error_messages = {
        "required": _("No file was submitted in field_title."),
        "invalid": _(
            "The submitted data was not a file. Check the encoding type on the"
            " form in field_title."
        ),
        "no_name": _("No filename could be determined in field_title."),
        "empty": _("The submitted file is empty in field_title."),
        "max_length": _(
            "Ensure this filename has at most {max_length} characters"
            " (it has {length}) in field_title."
        ),
    }


class CustomDateTimeField(DateTimeField):
    default_error_messages = {
        "invalid": _(
            "Datetime has wrong format. Use one of these formats instead: "
            "{format} in field_title."
        ),
        "date": _("field_title Expected a datetime but got a date."),
        "make_aware": _(
            'Invalid datetime for the timezone "{timezone} in field_title".'
        ),
        "overflow": _("Datetime value out of range in field_title."),
    }


class CustomUrlField(URLField):
    default_error_messages = {"invalid": _("Enter a valid URL.")}


class CustomJSONField(JSONField):
    default_error_messages = {
        "invalid": _("Value must be valid JSON in field_title."),
        "required": _("field_title field is required"),
    }


class UnvalidatedField(Field):
    default_error_messages = {
        "null": _("field_title may not be null."),
    }

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.allow_blank = True
        self.allow_null = False

    def to_internal_value(self, data):
        return data

    def to_representation(self, value):
        return value


def validate_serializers_message(errors):
    def extract_messages(error_obj, key=None):
        msgs = []
        if isinstance(error_obj, dict):
            for k, v in error_obj.items():
                msgs.extend(extract_messages(v, k))
        elif isinstance(error_obj, list):
            for item in error_obj:
                msgs.extend(extract_messages(item, key))
        elif isinstance(error_obj, str):
            replacement = key_map.get(key, key) if key else key
            if replacement:
                # Ensure replacement is a string to avoid TypeError
                replacement_str = (
                    str(replacement) if replacement is not None else ""
                )
                msgs.append(error_obj.replace("field_title", replacement_str))
            else:
                msgs.append(error_obj)
        return msgs

    msg = extract_messages(errors)
    return "|".join(msg)


class TenantScopedPrimaryKeyRelatedField(CustomPrimaryKeyRelatedField):
    # Narrows the candidate queryset to the acting user's tenant, so a pk
    # referencing another tenant's object fails validation as
    # does_not_exist (a 400) rather than silently binding a foreign row.
    def get_queryset(self):
        # No acting user means no context was passed: fail closed.
        queryset = super().get_queryset()
        user = acting_user(self.context)
        return queryset.for_user(user) if user else queryset.none()

1	from django.core.validators import EmailValidator	1✔
2
3	from utils.tenant_scoped_model import acting_user	1✔
4	from django.utils.translation import gettext_lazy as _	1✔
5	from rest_framework.fields import (	1✔
6	IntegerField,
7	ChoiceField,
8	CharField,
9	ImageField,
10	ListField,
11	BooleanField,
12	FloatField,
13	DecimalField,
14	URLField,
15	DateField,
16	MultipleChoiceField,
17	FileField,
18	DateTimeField,
19	JSONField,
20	Field,
21	)
22	from rest_framework.relations import PrimaryKeyRelatedField	1✔
23
24	key_map = {}	1✔
25
26
27	class CustomIntegerField(IntegerField):	1✔
28	default_error_messages = {	1✔
29	"invalid": _("A valid field_title is required."),
30	"max_value": _(
31	"Ensure field_title is less than or equal to {max_value}."
32	),
33	"min_value": _(
34	"Ensure field_title is greater than or equal to {min_value}."
35	),
36	"max_string_length": _("field_title value too large."),
37	"required": _("field_title is required."),
38	"null": _("field_title may not be null."),
39	}
40
41
42	class CustomCharField(CharField):	1✔
43	default_error_messages = {	1✔
44	"invalid": _("A valid field_title is required."),
45	"blank": _("field_title may not be blank."),
46	"max_length": _(
47	"Ensure field_title has no more than {max_length} characters."
48	),
49	"min_length": _(
50	"Ensure field_title has at least {min_length} characters."
51	),
52	"required": _("field_title is required."),
53	"null": _("field_title may not be null."),
54	}
55
56
57	class CustomChoiceField(ChoiceField):	1✔
58	default_error_messages = {	1✔
59	"invalid_choice": _('"{input}" is not a valid choice in field_title.'),
60	"required": _("field_title is required."),
61	"null": _("field_title may not be null."),
62	}
63
64
65	class CustomMultipleChoiceField(MultipleChoiceField):	1✔
66	default_error_messages = {	1✔
67	"invalid_choice": _('"{input}" is not a valid choice in field_title.'),
68	"not_a_list": _(
69	'Expected a list of items but got type "{input_type}"'
70	" in field_title."
71	),
72	"empty": _("This selection may not be empty in field_title."),
73	}
74
75
76	class CustomImageField(ImageField):	1✔
77	default_error_messages = {	1✔
78	"invalid_image": _(
79	"Upload a valid image. field_title you uploaded was either not "
80	"an image or a corrupted image."
81	),
82	"required": _("field_title is required."),
83	"null": _("field_title may not be null."),
84	}
85
86
87	class CustomEmailField(CustomCharField):	1✔
88	default_error_messages = {"invalid": _("Enter a valid email address.")}	1✔
89
90	def __init__(self, **kwargs):	1✔
91	super().__init__(**kwargs)	1✔
92	validator = EmailValidator(message=self.error_messages["invalid"])	1✔
93	self.validators.append(validator)	1✔
94
95
96	class CustomListField(ListField):	1✔
97	default_error_messages = {	1✔
98	"not_a_list": _(
99	"Expected a list of items in field_title but got type"
100	' "{input_type}".'
101	),
102	"empty": _("field_title may not be empty."),
103	"min_length": _(
104	"Ensure field_title has at least {min_length} elements."
105	),
106	"max_length": _(
107	"Ensure field_title has no more than {max_length} elements."
108	),
109	"required": _("field_title is required."),
110	"null": _("field_title may not be null."),
111	}
112
113
114	class CustomBooleanField(BooleanField):	1✔
115	default_error_messages = {	1✔
116	"invalid": _("Must be a valid field_title."),
117	"required": _("field_title is required."),
118	"null": _("field_title may not be null."),
119	}
120
121
122	class CustomFloatField(FloatField):	1✔
123	default_error_messages = {	1✔
124	"invalid": _("A valid field_title is required."),
125	"max_value": _(
126	"Ensure field_title is less than or equal to {max_value}."
127	),
128	"min_value": _(
129	"Ensure field_title is greater than or equal to {min_value}."
130	),
131	"max_string_length": _("field_title too large."),
132	"required": _("field_title is required."),
133	"null": _("field_title may not be null."),
134	}
135
136
137	class CustomDecimalField(DecimalField):	1✔
138	default_error_messages = {	1✔
139	"invalid": _("A field_title number is required."),
140	"max_value": _(
141	"Ensure field_title is less than or equal to {max_value}."
142	),
143	"min_value": _(
144	"Ensure field_title is greater than or equal to {min_value}."
145	),
146	"max_digits": _(
147	"Ensure that in field_title are no more than {max_digits} "
148	"digits in total."
149	),
150	"max_decimal_places": _(
151	"Ensure that in field_title are no more than {max_decimal_places} "
152	"decimal places."
153	),
154	"max_whole_digits": _(
155	"Ensure that in field_title are no more than {max_whole_digits} "
156	"digits before the "
157	"decimal point."
158	),
159	"max_string_length": _("String value too large in field_title."),
160	"required": _("field_title is required."),
161	"null": _("field_title may not be null."),
162	}
163
164
165	class CustomURLField(URLField):	1✔
166	default_error_messages = {	1✔
167	"invalid": _("Enter a valid URL in field_title."),
168	"required": _("field_title is required."),
169	"null": _("field_title may not be null."),
170	}
171
172
173	class CustomPrimaryKeyRelatedField(PrimaryKeyRelatedField):	1✔
174	default_error_messages = {	1✔
175	"required": _("field_title is required."),
176	"does_not_exist": _(
177	'Invalid pk "{pk_value}" - object does not exist.'
178	),
179	"incorrect_type": _(
180	"Incorrect type. Expected pk value, received {data_type}."
181	),
182	"null": _("field_title may not be null."),
183	}
184
185
186	class CustomDateField(DateField):	1✔
187	default_error_messages = {	1✔
188	"required": _("field_title is required."),
189	"invalid": _(
190	"Date has wrong format. Use one of these formats instead:"
191	" {format}."
192	),
193	"datetime": _("Expected a date but got a datetime."),
194	"null": _("field_title may not be null."),
195	}
196
197
198	class CustomFileField(FileField):	1✔
199	default_error_messages = {	1✔
200	"required": _("No file was submitted in field_title."),
201	"invalid": _(
202	"The submitted data was not a file. Check the encoding type on the"
203	" form in field_title."
204	),
205	"no_name": _("No filename could be determined in field_title."),
206	"empty": _("The submitted file is empty in field_title."),
207	"max_length": _(
208	"Ensure this filename has at most {max_length} characters"
209	" (it has {length}) in field_title."
210	),
211	}
212
213
214	class CustomDateTimeField(DateTimeField):	1✔
215	default_error_messages = {	1✔
216	"invalid": _(
217	"Datetime has wrong format. Use one of these formats instead: "
218	"{format} in field_title."
219	),
220	"date": _("field_title Expected a datetime but got a date."),
221	"make_aware": _(
222	'Invalid datetime for the timezone "{timezone} in field_title".'
223	),
224	"overflow": _("Datetime value out of range in field_title."),
225	}
226
227
228	class CustomUrlField(URLField):	1✔
229	default_error_messages = {"invalid": _("Enter a valid URL.")}	1✔
230
231
232	class CustomJSONField(JSONField):	1✔
233	default_error_messages = {	1✔
234	"invalid": _("Value must be valid JSON in field_title."),
235	"required": _("field_title field is required"),
236	}
237
238
239	class UnvalidatedField(Field):	1✔
240	default_error_messages = {	1✔
241	"null": _("field_title may not be null."),
242	}
243
244	def __init__(self, **kwargs):	1✔
245	super().__init__(**kwargs)	1✔
246	self.allow_blank = True	1✔
247	self.allow_null = False	1✔
248
249	def to_internal_value(self, data):	1✔
250	return data	1✔
251
252	def to_representation(self, value):	1✔
253	return value	×
254
255
256	def validate_serializers_message(errors):	1✔
257	def extract_messages(error_obj, key=None):	1✔
258	msgs = []	1✔
259	if isinstance(error_obj, dict):	1✔
260	for k, v in error_obj.items():	1✔
261	msgs.extend(extract_messages(v, k))	1✔
262	elif isinstance(error_obj, list):	1✔
263	for item in error_obj:	1✔
264	msgs.extend(extract_messages(item, key))	1✔
265	elif isinstance(error_obj, str):	1!
266	replacement = key_map.get(key, key) if key else key	1✔
267	if replacement:	1✔
268	# Ensure replacement is a string to avoid TypeError
269	replacement_str = (	1✔
270	str(replacement) if replacement is not None else ""
271	)
272	msgs.append(error_obj.replace("field_title", replacement_str))	1✔
273	else:
274	msgs.append(error_obj)	1✔
275	return msgs	1✔
276
277	msg = extract_messages(errors)	1✔
278	return "\|".join(msg)	1✔
279
280
281	class TenantScopedPrimaryKeyRelatedField(CustomPrimaryKeyRelatedField):	1✔
282	# Narrows the candidate queryset to the acting user's tenant, so a pk
283	# referencing another tenant's object fails validation as
284	# does_not_exist (a 400) rather than silently binding a foreign row.
285	def get_queryset(self):	1✔
286	# No acting user means no context was passed: fail closed.
287	queryset = super().get_queryset()	1✔
288	user = acting_user(self.context)	1✔
289	return queryset.for_user(user) if user else queryset.none()	1✔

akvo / akvo-mis / #773

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous