fix(security): isolate template rendering with hard resource limits

This commit is contained in:
2026-09-08 07:47:18 +02:00
parent 1f61464fd6
commit 1b1e3e0f92
6 changed files with 229 additions and 1 deletions
+31 -1
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import hashlib
import unittest
from dataclasses import replace
from unittest.mock import patch
from govoplan_core.auth import ApiPrincipal
@@ -19,6 +20,7 @@ from govoplan_core.core.templates import (
)
from govoplan_core.db.base import Base
from govoplan_core.db.session import configure_database, reset_database
from govoplan_templates.backend import rendering
from govoplan_templates.backend.capabilities import (
SqlTemplateCatalog,
SqlTemplateContentLibrary,
@@ -231,7 +233,8 @@ class TemplateServiceTests(unittest.TestCase):
idempotency_key="campaign-1:postal-output-1",
)
first = render_template(session, principal(), registry=_Registry(), request=request)
second = render_template(session, principal(), registry=_Registry(), request=request)
with patch.object(rendering, "run_bounded_operation", side_effect=AssertionError("idempotent render started a worker")):
second = render_template(session, principal(), registry=_Registry(), request=request)
session.commit()
self.assertEqual(first.render_id, second.render_id)
@@ -243,6 +246,33 @@ class TemplateServiceTests(unittest.TestCase):
self.assertIn(b"Ada", first.payload)
self.assertIn(b"Grace", first.payload)
def test_real_worker_timeout_writes_neither_artifact_nor_render_evidence(self) -> None:
store = _ArtifactStore()
with self.database.session() as session:
item, _ = create_template(session, principal(), payload())
request = TemplateRenderRequest(
template_id=item.id,
usage="campaign.postal",
items=({"name": "Ada", "postal": {"address": "Street 1"}},),
persist_to_files=True,
)
limits = replace(rendering.RENDER_PROCESS_LIMITS, wall_seconds=0.001)
with patch.object(rendering, "RENDER_PROCESS_LIMITS", limits):
with self.assertRaisesRegex(TemplateRenderError, "timeout"):
render_template(session, principal(), registry=_Registry(store), request=request)
self.assertIsNone(store.request)
self.assertEqual([], list_renders(session, principal()))
def test_incompatible_input_is_rejected_before_worker_admission(self) -> None:
with self.database.session() as session:
item, _ = create_template(session, principal(), payload())
request = TemplateRenderRequest(
template_id=item.id, usage="campaign.postal", items=({"name": "Missing address"},),
)
with patch.object(rendering, "run_bounded_operation", side_effect=AssertionError("incompatible render started a worker")):
with self.assertRaises(TemplateCompatibilityError):
render_template(session, principal(), registry=_Registry(), request=request)
def test_bounded_render_history_and_payload_are_owner_scoped(self) -> None:
owner = principal(admin=False)
other = principal(account_id="account-2", admin=False)