Commit b4c2211d authored by Mickaël Desfrênes's avatar Mickaël Desfrênes
Browse files

Prevent snapshot restore from broadening stale object permissions 

Permissions referencing hard-deleted objects can remain in project snapshots.
Restoration previously mapped their missing targets to object_pk=None, turning
scoped permissions into class-wide grants or denials and potentially exposing
unrelated project data.

Skip permissions whose targets cannot be remapped while preserving valid scoped
permissions and intentional class-wide permissions. Report skipped entries in
one warning with at most five examples, without adding database queries.

Add round-trip regression coverage for stale grants and denials, valid target ID
remapping, preservation of class-wide permissions, and bounded warning output.

Validation: all 18 ProjectSnapshotTestCase tests passed; Ruff and git diff
--check passed.
parent 5640d5f4

src/resources/snapshots.py

+22 −1
Changes for src/resources/snapshots.py: 22 added lines, 1 removed line.
Original line number Diff line number Diff line 
import json

import logging

import shutil

from contextlib import contextmanager

from pathlib import Path
@@ -33,6 +34,8 @@ from resources.models import ( 
    resource_post_save,

)



logger = logging.getLogger(__name__)



SNAPSHOT_FORMAT = "jama.project_snapshot.v1"

SECTION_NAMES = [

    "project",
@@ -844,6 +847,8 @@ def restore_project_snapshot( 
                }



                def object_permission_values():

                    skipped_count = 0

                    skipped_examples = []

                    for row in _read_jsonl(

                        snapshot_dir.joinpath("object_permissions.ndjson")

                    ):
@@ -853,7 +858,15 @@ def restore_project_snapshot( 
                            and row["object_class"] in object_class_maps

                        ):

                            sid_name, sid_map = object_class_maps[row["object_class"]]

                            object_pk = sid_map.get(_sid(sid_name, object_pk))

                            restored_object_pk = sid_map.get(_sid(sid_name, object_pk))

                            if restored_object_pk is None:

                                skipped_count += 1

                                if len(skipped_examples) < 5:

                                    skipped_examples.append(

                                        f"{row['object_class']}({object_pk}) for {row['role']}"

                                    )

                                continue

                            object_pk = restored_object_pk

                        yield ObjectPermission(

                            role_id=roles[row["role"]],

                            object_class=row["object_class"],
@@ -863,6 +876,14 @@ def restore_project_snapshot( 
                            object_update=row.get("object_update") or False,

                            object_delete=row.get("object_delete") or False,

                        )

                    if skipped_count:

                        logger.warning(

                            "Skipped %d scoped permission(s) with missing targets "

                            "while restoring project %s; examples: %s",

                            skipped_count,

                            project.label,

                            ", ".join(skipped_examples),

                        )



                _bulk_create_stream(

                    ObjectPermission,

src/resources/tests.py

+122 −0
Changes for src/resources/tests.py: 122 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -1130,6 +1130,128 @@ class ProjectSnapshotTestCase(TestCase): 
        self.assertEqual(resource_rows[0]["title"], "text file")

        self.assertEqual(resource_rows[0]["kind"], "file")



    def test_restore_skips_stale_scoped_permissions(self):

        ObjectPermission.objects.filter(role=self.role).delete()

        stale_resource = models.Resource.objects.create(

            ptr_project=self.project, title="removed resource"

        )

        stale_collection = models.Collection.objects.create(

            project=self.project, parent=self.root, title="removed collection"

        )

        stale_file = File.objects.create(

            project=self.project,

            title="removed file",

            original_name="removed.txt",

            hash="c" * 64,

            file_type=self.file_type,

            size=0,

        )

        for object_class, target in (

            ("resource", stale_resource),

            ("collection", stale_collection),

            ("file", stale_file),

        ):

            for allowed in (True, False):

                ObjectPermission.objects.create(

                    role=self.role,

                    object_class=object_class,

                    object_pk=target.pk,

                    object_read=allowed,

                    object_update=allowed,

                )

            target.delete()



        with TemporaryDirectory() as snapshot_dir:

            manifest = snapshots.export_project_snapshot(self.project.pk, snapshot_dir)

            self.assertEqual(manifest["counts"]["object_permissions"], 6)

            self.project.delete()

            with self.assertLogs("resources.snapshots", level="WARNING") as logs:

                restored = snapshots.restore_project_snapshot(

                    snapshot_dir, "snapshot restored"

                )



        self.assertEqual(len(logs.records), 1)

        self.assertIn("Skipped 6 scoped permission(s)", logs.output[0])

        self.assertEqual(len(logs.records[0].args[-1].split(", ")), 5)

        self.assertFalse(

            ObjectPermission.objects.filter(role__project=restored).exists()

        )

        restored_resource = models.Resource.objects.get(

            ptr_project=restored, title="plain resource"

        )

        restored_collection = models.Collection.objects.get(

            project=restored, title="child"

        )

        restored_access = UserAccess(self.user, restored)

        self.assertTrue(restored_access.has_project_access())

        self.assertFalse(restored_access.can_read(restored_resource))

        self.assertFalse(restored_access.can_update(restored_resource))

        self.assertFalse(restored_access.can_read(restored_collection))



    def test_restore_remaps_valid_scoped_permissions(self):

        ObjectPermission.objects.filter(role=self.role).delete()

        for object_class, target in (

            ("resource", self.resource),

            ("collection", self.child),

            ("file", self.file),

        ):

            ObjectPermission.objects.create(

                role=self.role,

                object_class=object_class,

                object_pk=target.pk,

                object_read=True,

                object_update=True,

            )



        with TemporaryDirectory() as snapshot_dir:

            snapshots.export_project_snapshot(self.project.pk, snapshot_dir)

            self.project.delete()

            restored = snapshots.restore_project_snapshot(

                snapshot_dir, "snapshot restored"

            )



        restored_resource = models.Resource.objects.get(

            ptr_project=restored, title="plain resource"

        )

        restored_collection = models.Collection.objects.get(

            project=restored, title="child"

        )

        restored_file = File.objects.get(project=restored)

        for object_class, target in (

            ("resource", restored_resource),

            ("collection", restored_collection),

            ("file", restored_file),

        ):

            permission = ObjectPermission.objects.get(

                role__project=restored, object_class=object_class

            )

            self.assertEqual(permission.object_pk, target.pk)

            self.assertTrue(permission.object_read)

            self.assertTrue(permission.object_update)

        restored_access = UserAccess(self.user, restored)

        self.assertTrue(restored_access.can_read(restored_resource))

        self.assertTrue(restored_access.can_update(restored_collection))



    def test_restore_preserves_class_wide_permissions(self):

        with TemporaryDirectory() as snapshot_dir:

            snapshots.export_project_snapshot(self.project.pk, snapshot_dir)

            self.project.delete()

            restored = snapshots.restore_project_snapshot(

                snapshot_dir, "snapshot restored"

            )



        permission = ObjectPermission.objects.get(role__project=restored)

        self.assertEqual(permission.object_class, "resource")

        self.assertIsNone(permission.object_pk)

        self.assertTrue(permission.object_read)

        self.assertFalse(permission.object_update)

        restored_resource = models.Resource.objects.get(

            ptr_project=restored, title="plain resource"

        )

        restored_access = UserAccess(self.user, restored)

        self.assertTrue(restored_access.can_read(restored_resource))

        self.assertFalse(restored_access.can_update(restored_resource))



    def test_restore_without_file_blobs_does_not_enqueue_media_tasks(self):

        content = b"blob"

        file_hash = hashlib.sha256(content).hexdigest()