Coverage for backend/django/core/validation.py: 86%
241 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1import contextvars
2from contextlib import contextmanager
3from django.apps import apps
4from functools import wraps
5from authentication.user.models import User
6from core.viewset import ModelViewSet, ReadOnlyModelViewSet
7from django.http import JsonResponse
8from django.urls.resolvers import URLPattern, URLResolver
9from rest_framework.exceptions import APIException, NotFound, PermissionDenied, ValidationError
10from rest_framework.permissions import SAFE_METHODS
11from rest_framework.request import Request
12from rest_framework.viewsets import ModelViewSet as MDVS
13from typing import TypedDict
15def sanitize_flowsheet_id(flowsheet_id):
16 """
17 Ensure the client has provided a valid flowsheet id (positive integer)
18 """
19 try:
20 flowsheet_id = int(flowsheet_id)
21 if not flowsheet_id:
22 raise ValidationError("Invalid flowsheet id")
23 if flowsheet_id < 1:
24 raise ValidationError("Invalid flowsheet id")
25 except:
26 raise ValidationError("Invalid flowsheet id")
28 return flowsheet_id
31def reject_historical_revision_write(request: Request) -> None:
32 """Reject unsafe API requests that declare immutable revision read scope."""
34 if (
35 request.method not in SAFE_METHODS
36 and request.GET.get("revision") is not None
37 ):
38 raise PermissionDenied("Historical revisions are read-only.")
41def _api_exception_json_response(exc: APIException) -> JsonResponse:
42 """Render validation raised outside DRF dispatch with DRF's response shape."""
44 detail = exc.detail
45 if isinstance(detail, dict): 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true
46 data = detail
47 safe = True
48 elif isinstance(detail, list):
49 data = detail
50 safe = False
51 else:
52 data = {"detail": detail}
53 safe = True
54 return JsonResponse(data, status=exc.status_code, safe=safe)
57def api_view_validate(view_func):
58 """
59 Decorator for every api_view to enforce access control
60 """
62 @wraps(view_func)
63 def _wrapped_view(request: Request, *args, **kwargs):
64 try:
65 reject_historical_revision_write(request)
66 flowsheet_id = request.GET.get("flowsheet")
67 revision_state_id = request.GET.get("revision")
68 user = request.user
69 flowsheet_id = sanitize_flowsheet_id(flowsheet_id)
71 with flowsheet_context(
72 flowsheet_id,
73 user,
74 revision_state_id=revision_state_id,
75 write_intent=request.method not in SAFE_METHODS,
76 ):
77 return view_func(request, *args, **kwargs)
78 except APIException as exc:
79 # @api_view_validate wraps the @api_view callable, so access errors
80 # raised here occur before DRF's dispatch exception handler runs.
81 return _api_exception_json_response(exc)
82 _wrapped_view._is_api_view_validated = True
83 return _wrapped_view
85def api_view_ignore_access_control(view_func):
86 """
87 Decorator for api_view to ignore access control. This is not for
88 general use. Attach only to internal endpoints/handlers (such as
89 for Dapr to invoke).
90 """
92 @wraps(view_func)
93 def _wrapped_view(request: Request, *args, **kwargs):
94 return view_func(request, *args, **kwargs)
96 _wrapped_view.ignore_access_control = True
97 return _wrapped_view
101################## Context manager ##################
102class FlowsheetContext(TypedDict):
103 flowsheet: int
104 flowsheet_state: int
105 historical_read: bool
106 user: User
107 has_access: bool | None # Backward-compatible alias for has_read_access.
108 has_read_access: bool | None
109 has_write_access: bool | None
110 write_intent: bool # Lets manager/queryset code distinguish read vs write views.
111 enforce_write_access: bool # Strictly reject unsafe custom actions before their body runs.
112 bypass_write_checks: bool # Temporary escape hatch for explicitly validated internal flows.
114flowsheet_ctx = contextvars.ContextVar[FlowsheetContext | None]("flowsheet", default=None)
115write_check_bypass_ctx = contextvars.ContextVar("write_check_bypass", default=False)
117@contextmanager
118def flowsheet_context(
119 flowsheet: int,
120 user: User,
121 revision_state_id: int | str | None = None,
122 write_intent: bool = False,
123 enforce_write_access: bool = False,
124):
125 """
126 Store the active flowsheet, request user, and write-intent flag in a
127 request-local context so managers/querysets can enforce access control
128 without every model method needing the request passed through explicitly.
129 """
131 from core.auxiliary.models.Flowsheet import Flowsheet
132 from core.auxiliary.models.FlowsheetState import (
133 FlowsheetState,
134 FlowsheetStateRole,
135 )
137 historical_read = revision_state_id is not None
138 historical_access_state = None
139 if historical_read:
140 if write_intent:
141 raise PermissionDenied("Historical revisions are read-only.")
143 # Resolve stable-resource access first so revision IDs cannot be used as
144 # an existence oracle by users who cannot read the flowsheet.
145 from core.managers import get_flowsheet_access
147 historical_access_state = get_flowsheet_access(
148 user=user,
149 flowsheet_id=flowsheet,
150 )
151 if not historical_access_state.has_read_access:
152 raise PermissionDenied("User does not have access to this flowsheet.")
154 try:
155 normalized_revision_state_id = int(revision_state_id)
156 except (TypeError, ValueError) as exc:
157 raise NotFound("Revision not found.") from exc
158 if normalized_revision_state_id < 1: 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true
159 raise NotFound("Revision not found.")
160 flowsheet_state = (
161 FlowsheetState.objects.filter(
162 pk=normalized_revision_state_id,
163 flowsheet_id=flowsheet,
164 role=FlowsheetStateRole.REVISION,
165 )
166 .values_list("pk", flat=True)
167 .first()
168 )
169 if flowsheet_state is None:
170 raise NotFound("Revision not found.")
171 else:
172 flowsheet_state = (
173 Flowsheet.objects
174 .filter(pk=flowsheet)
175 .values_list("current_state_id", flat=True)
176 .first()
177 )
178 if flowsheet_state is None:
179 raise ValidationError("Flowsheet has no active working state.")
181 data = FlowsheetContext({
182 "flowsheet": flowsheet,
183 "flowsheet_state": flowsheet_state,
184 "historical_read": historical_read,
185 "user": user,
186 "has_access": (
187 historical_access_state.has_read_access
188 if historical_access_state is not None
189 else None
190 ),
191 "has_read_access": (
192 historical_access_state.has_read_access
193 if historical_access_state is not None
194 else None
195 ),
196 "has_write_access": (
197 historical_access_state.has_write_access
198 if historical_access_state is not None
199 else None
200 ),
201 "write_intent": write_intent,
202 "enforce_write_access": enforce_write_access,
203 "bypass_write_checks": False,
204 })
205 token = flowsheet_ctx.set(data)
207 try:
208 if enforce_write_access:
209 from core.managers import get_flowsheet_access
211 access_state = get_flowsheet_access(user=user, flowsheet_id=flowsheet)
212 if not access_state.has_write_access:
213 raise PermissionDenied("User does not have write access to this flowsheet.")
215 # Allow the view to execute with the context set
216 yield
217 finally:
218 try:
219 flowsheet_ctx.reset(token)
220 except ValueError:
221 # This is to reset the context if the server throws an error while processing the request
222 # So that it doesn't leak to the next request
223 flowsheet_ctx.set(None)
225def get_current_flowsheet():
226 """
227 Get the current flowsheet and user id from the context in format
228 {
229 "flowsheet": flowsheet_id,
230 "flowsheet_state": selected working or historical state id,
231 "historical_read": whether a validated revision was selected,
232 "user": User,
233 "has_access": has_access
234 }
235 """
236 return flowsheet_ctx.get()
239def write_access_checks_are_bypassed() -> bool:
240 """Return whether a narrowly scoped internal state mutation is active."""
242 return write_check_bypass_ctx.get()
245def cache_result(has_access: bool = False):
246 """
247 Cache the result of the flowsheet context
248 """
249 data = flowsheet_ctx.get()
250 if not data:
251 return
253 data["has_access"] = has_access
254 data["has_read_access"] = has_access
255 flowsheet_ctx.set(data)
258def cache_access_result(
259 *,
260 has_read_access: bool | None = None,
261 has_write_access: bool | None = None,
262):
263 """
264 Cache read/write flowsheet access in request context.
265 """
266 data = flowsheet_ctx.get()
267 if not data: 267 ↛ 268line 267 didn't jump to line 268 because the condition on line 267 was never true
268 return
270 if has_read_access is not None: 270 ↛ 274line 270 didn't jump to line 274 because the condition on line 270 was always true
271 data["has_read_access"] = has_read_access
272 data["has_access"] = has_read_access
274 if has_write_access is not None: 274 ↛ 277line 274 didn't jump to line 277 because the condition on line 274 was always true
275 data["has_write_access"] = has_write_access
277 flowsheet_ctx.set(data)
280@contextmanager
281def bypass_write_access_checks():
282 """
283 Temporarily bypass manager/queryset write checks for special internal flows
284 that already perform their own explicit access validation.
285 """
286 data = flowsheet_ctx.get()
287 previous_value = data.get("bypass_write_checks", False) if data else False
288 if data:
289 data["bypass_write_checks"] = True
290 flowsheet_ctx.set(data)
291 token = write_check_bypass_ctx.set(True)
293 try:
294 yield
295 finally:
296 write_check_bypass_ctx.reset(token)
297 if data:
298 data = flowsheet_ctx.get() or data
299 data["bypass_write_checks"] = previous_value
300 flowsheet_ctx.set(data)
302################## Router and urlpattern validation ##################
305def validate_router(router):
306 """
307 Validate that every registered viewset is a subclass of ModelViewSet
308 """
309 # Projects and project folders are explicitly owner-scoped resources rather
310 # than flowsheet-scoped models, so their focused viewsets enforce access at
311 # that domain boundary instead of inheriting the flowsheet ModelViewSet.
312 exclude_list = [
313 "flowsheets",
314 "flowsheetTemplates",
315 "compounds",
316 "projects",
317 "project-folders",
318 ]
319 viewsets = [ModelViewSet, ReadOnlyModelViewSet]
320 for (prefix, viewset, basename) in router.registry:
321 if prefix in exclude_list:
322 continue
324 if not any(issubclass(viewset, vs) for vs in viewsets): 324 ↛ 325line 324 didn't jump to line 325 because the condition on line 324 was never true
325 raise Exception(f"ModelViewSet (from core.viewset) is not being inherited at {prefix}!")
327 if getattr(viewset, "get_queryset", None) == getattr(MDVS, "get_queryset"): 327 ↛ 328line 327 didn't jump to line 328 because the condition on line 327 was never true
328 raise Exception(f"get_queryset is not being overridden at {prefix}! Please override get_queryset method to provide the queryset")
330 if getattr(viewset, "queryset", None) != None: 330 ↛ 331line 330 didn't jump to line 331 because the condition on line 330 was never true
331 raise Exception(f"Please remove queryset from {prefix} viewset and create a get_queryset method instead. \
332 Avoid creating the queryset attribute to enforce access control.")
335def validate_urlpatterns(urlpatterns):
336 """
337 Validate that every view in urlpatterns is decorated with api_view_validate
338 """
339 all_views = extract_views_from_urlpatterns(urlpatterns)
340 for path, view in all_views:
341 if not hasattr(view, '_is_api_view_validated'): 341 ↛ 342line 341 didn't jump to line 342 because the condition on line 341 was never true
342 raise Exception(f"api_view_validate decorator (from core.validation) is not being used at {path}!!!")
345def extract_views_from_urlpatterns(urlpatterns, base_path=''):
346 """
347 Recursively extracts views from urlpatterns.
348 """
349 views = []
350 for pattern in urlpatterns:
351 if isinstance(pattern, URLPattern): 351 ↛ 354line 351 didn't jump to line 354 because the condition on line 351 was always true
352 path = base_path + str(pattern.pattern)
353 views.append((path, pattern.callback))
354 elif isinstance(pattern, URLResolver): # nested patterns (like routers)
355 nested_path = base_path + str(pattern.pattern)
356 views.extend(extract_views_from_urlpatterns(
357 pattern.url_patterns, nested_path))
358 return views
361# Check that scoped models declare state, identity, or project ownership.
362def validate_models():
363 from core.managers import (
364 AccessControlManager,
365 IdentityAccessControlManager,
366 ProjectAccessControlManager,
367 SoftDeleteManager,
368 UserAccessControlManager,
369 )
370 exclude_models = [
371 'User',
372 'Permission',
373 'Group',
374 'ContentType',
375 'Flowsheet',
376 'FlowsheetState',
377 'AccessTable',
378 'Session',
379 'TaskMeta',
380 'Project',
381 'CostIndexSeries',
382 'CostIndexValue',
383 'EconomicsDefaultRate',
384 'EconomicsLangFactorDefault',
385 'ProjectFolder',
386 ]
388 models = apps.get_models()
391 for model in models:
392 model_name = model.__name__
393 module_name = model.__module__
394 # Skip silk models
395 if module_name.startswith('silk'): 395 ↛ 396line 395 didn't jump to line 396 because the condition on line 395 was never true
396 continue
397 objects = model.objects
398 if model_name in exclude_models:
399 continue
401 if 'flowsheetOwner' in [field.name for field in model._meta.get_fields()]: 401 ↛ 402line 401 didn't jump to line 402 because the condition on line 401 was never true
402 raise ValueError("To enforce access control, `flowsheetOwner` should not be used. Please rename to `flowsheet` instead")
404 has_scope = False
406 assert (
407 isinstance(
408 objects,
409 (
410 AccessControlManager,
411 IdentityAccessControlManager,
412 ProjectAccessControlManager,
413 SoftDeleteManager,
414 UserAccessControlManager,
415 ),
416 )
417 ), f"Model {model_name} does not use an approved access-control manager."
419 if hasattr(model, 'flowsheet_state') and isinstance(objects, AccessControlManager):
420 has_scope = True
421 if hasattr(model, 'flowsheet') and isinstance(objects, IdentityAccessControlManager):
422 has_scope = True
423 if hasattr(model, 'project') and isinstance(objects, ProjectAccessControlManager):
424 has_scope = True
425 if hasattr(model, 'owner') and isinstance(objects, UserAccessControlManager):
426 has_scope = True
428 if hasattr(model, 'flowsheetOwner'): 428 ↛ 429line 428 didn't jump to line 429 because the condition on line 428 was never true
429 raise ValueError("To enforce access control, `flowsheetOwner` should not be used. Please rename to `flowsheet` instead")
431 assert has_scope, f"Model {model_name} does not declare state, identity, or project scope."
433def validate_routers():
434 import os
435 import ast
437 IGNORED_DIRS = {"site-packages"}
438 IGNORED_FILES = {
439 os.path.normpath("./authentication/routers.py")
440 }
442 def is_ignored(path: str) -> bool:
443 norm = os.path.normpath(path)
444 parts = norm.split(os.sep)
446 # Ignore specific directories
447 if any(part in IGNORED_DIRS for part in parts):
448 return True
450 # Ignore specific file paths
451 if norm in IGNORED_FILES:
452 return True
454 return False
456 issues = []
458 for root, _, files in os.walk("./"):
459 if is_ignored(root):
460 continue
462 for file in files:
463 if file == "routers.py":
464 path = os.path.join(root, file)
466 if is_ignored(path):
467 continue
469 with open(path, "r", encoding="utf-8") as f:
470 source = f.read()
472 tree = ast.parse(source)
474 found = False
476 # inspect AST for validate_router(...) calls
477 for node in ast.walk(tree): 477 ↛ 493line 477 didn't jump to line 493 because the loop on line 477 didn't complete
478 if isinstance(node, ast.Call):
479 func = node.func
481 # Extract function name
482 if isinstance(func, ast.Name):
483 name = func.id
484 elif isinstance(func, ast.Attribute): 484 ↛ 487line 484 didn't jump to line 487 because the condition on line 484 was always true
485 name = func.attr
486 else:
487 continue
489 if name == "validate_router":
490 found = True
491 break
493 if not found: 493 ↛ 494line 493 didn't jump to line 494 because the condition on line 493 was never true
494 issues.append(path)
497 if issues: 497 ↛ 498line 497 didn't jump to line 498 because the condition on line 497 was never true
498 issue_list = "\n".join(issues)
499 raise Exception(f"The following routers.py files are missing validate_router(...) calls:\n{issue_list}")