Fail open when View recovery surfaces disappear

This commit is contained in:
2026-07-28 21:12:29 +02:00
parent e14432db73
commit ca615df36e
2 changed files with 91 additions and 4 deletions

View File

@@ -942,7 +942,8 @@ def resolve_effective_view(
diagnostics: list[ViewDiagnostic] = []
if selected is not None and catalogue is not None:
assignment, revision = selected
known_surfaces = {surface.id: surface for surface in catalogue}
current_catalogue = tuple(catalogue)
known_surfaces = {surface.id: surface for surface in current_catalogue}
stale_ids = sorted(set(revision.visible_surface_ids) - set(known_surfaces))
if stale_ids:
diagnostics.append(
@@ -956,14 +957,58 @@ def resolve_effective_view(
surface_ids=tuple(stale_ids),
)
)
if locked:
try:
lockout_ids = lockout_required_surface_ids(
current_catalogue,
assignment_scope_type=assignment.scope_type,
)
unavailable_escape_ids = sorted(
lockout_ids - set(revision.visible_surface_ids)
)
lockout_message = (
"The required View no longer retains all administration "
"escape surfaces."
if unavailable_escape_ids
else ""
)
except ViewsValidationError as exc:
unavailable_escape_ids = ()
lockout_message = str(exc)
if lockout_message:
diagnostics.append(
ViewDiagnostic(
severity="error",
code="view.lockout_escape_unavailable",
message=(
f"{lockout_message} The full authorized interface "
"is shown until the assignment is repaired."
),
surface_ids=tuple(unavailable_escape_ids),
)
)
provenance.append(
{
"source": "invalid_view_fallback",
"scope_type": assignment.scope_type,
"scope_id": assignment.scope_id,
"detail": (
"Required View administration escape is unavailable "
f"for revision {revision.id}"
),
}
)
selected = None
locked = False
known_visible = [
known_surfaces[surface_id]
for surface_id in revision.visible_surface_ids
if surface_id in known_surfaces
]
if not any(
surface.kind == "navigation" for surface in known_visible
) or not any(surface.kind == "route" for surface in known_visible):
if selected is not None and (
not any(surface.kind == "navigation" for surface in known_visible)
or not any(surface.kind == "route" for surface in known_visible)
):
diagnostics.append(
ViewDiagnostic(
severity="error",

View File

@@ -310,6 +310,48 @@ class ViewsServiceTests(unittest.TestCase):
state.effective.provenance[0]["source"],
)
def test_required_view_falls_back_when_admin_escape_surface_is_removed(
self,
) -> None:
definition = self.create_published_definition(
name="Required task",
scope_type="system",
visible_surface_ids=lockout_safe_surface_ids(),
)
create_assignment(
self.session,
tenant_id="tenant-1",
scope_type="system",
scope_id=None,
definition=definition,
revision_id=None,
mode="required",
priority=0,
is_active=True,
metadata={},
catalogue=self.catalogue,
actor_id="account-admin",
)
catalogue_without_admin_route = tuple(
surface
for surface in self.catalogue
if surface.id != "access.route.admin"
)
state = resolve_effective_view(
self.session,
tenant_id="tenant-1",
account_id="account-user",
catalogue=catalogue_without_admin_route,
)
self.assertIsNone(state.effective.view_id)
self.assertFalse(state.effective.locked)
self.assertIn(
"view.lockout_escape_unavailable",
{diagnostic.code for diagnostic in state.diagnostics},
)
def test_required_assignment_rejects_revision_without_escape_surfaces(self) -> None:
definition = self.create_published_definition(name="Unsafe task")