/files: implemented /lock, /unlock endpoints

sachet/server/files/views.py

@@ -18,11 +18,15 @@ class FilesMetadataAPI(ModelAPI):
     @auth_required(required_permissions=(Permissions.MODIFY,), allow_anonymous=True)
     def patch(self, share_id, auth_user=None):
         share = Share.query.filter_by(share_id=share_id).first()
+        if share.locked:
+            return jsonify({"status": "fail", "message": "This share is locked."}), 423
         return super().patch(share)
 
     @auth_required(required_permissions=(Permissions.MODIFY,), allow_anonymous=True)
     def put(self, share_id, auth_user=None):
         share = Share.query.filter_by(share_id=share_id).first()
+        if share.locked:
+            return jsonify({"status": "fail", "message": "This share is locked."}), 423
         return super().put(share)
 
     @auth_required(required_permissions=(Permissions.DELETE,), allow_anonymous=True)
@@ -32,6 +36,8 @@ class FilesMetadataAPI(ModelAPI):
         except ValueError:
             return jsonify(dict(status="fail", message=f"Invalid ID: '{share_id}'."))
         share = Share.query.filter_by(share_id=share_id).first()
+        if share.locked:
+            return jsonify({"status": "fail", "message": "This share is locked."}), 423
         return super().delete(share)
 
 
@@ -117,6 +123,11 @@ class FileContentAPI(ModelAPI):
                 jsonify({"status": "fail", "message": "This share does not exist."})
             ), 404
 
+        if share.locked:
+            return (
+                jsonify({"status": "fail", "message": "This share is locked."})
+            ), 423
+
         if auth_user != share.owner:
             return (
                 jsonify(
@@ -182,3 +193,47 @@ files_blueprint.add_url_rule(
     view_func=FileContentAPI.as_view("files_content_api"),
     methods=["POST", "PUT", "GET"],
 )
+
+
+class FileLockAPI(ModelAPI):
+    @auth_required(required_permissions=(Permissions.LOCK,), allow_anonymous=True)
+    def post(self, share_id, auth_user=None):
+        share = Share.query.filter_by(share_id=uuid.UUID(share_id)).first()
+        if not share:
+            return (
+                jsonify({"status": "fail", "message": "This share does not exist."})
+            ), 404
+
+        share.locked = True
+        db.session.commit()
+
+        return jsonify({"status": "success", "message": "Share has been locked."})
+
+
+files_blueprint.add_url_rule(
+    "/files/<share_id>/lock",
+    view_func=FileLockAPI.as_view("files_lock_api"),
+    methods=["POST"],
+)
+
+
+class FileUnlockAPI(ModelAPI):
+    @auth_required(required_permissions=(Permissions.LOCK,), allow_anonymous=True)
+    def post(self, share_id, auth_user=None):
+        share = Share.query.filter_by(share_id=uuid.UUID(share_id)).first()
+        if not share:
+            return (
+                jsonify({"status": "fail", "message": "This share does not exist."})
+            ), 404
+
+        share.locked = False
+        db.session.commit()
+
+        return jsonify({"status": "success", "message": "Share has been unlocked."})
+
+
+files_blueprint.add_url_rule(
+    "/files/<share_id>/unlock",
+    view_func=FileUnlockAPI.as_view("files_unlock_api"),
+    methods=["POST"],
+)

sachet/server/models.py

@@ -220,6 +220,8 @@ class Share(db.Model):
     initialized : bool
         Since only the metadata is uploaded first, this switches to True when
         the real data is uploaded.
+    locked : bool
+        Locks modification and deletion of this share.
     create_date : DateTime
         Time the share was created (not initialized.)
     file_name : str
@@ -242,12 +244,13 @@ class Share(db.Model):
     owner = db.relationship("User", backref=db.backref("owner"))
 
     initialized = db.Column(db.Boolean, nullable=False, default=False)
+    locked = db.Column(db.Boolean, nullable=False, default=False)
 
     create_date = db.Column(db.DateTime, nullable=False)
 
     file_name = db.Column(db.String, nullable=False)
 
-    def __init__(self, owner_name=None, file_name=None):
+    def __init__(self, owner_name=None, file_name=None, locked=False):
         self.owner = User.query.filter_by(username=owner_name).first()
         if self.owner:
             self.owner_name = self.owner.username
@@ -259,6 +262,8 @@ class Share(db.Model):
         else:
             self.file_name = str(self.share_id)
 
+        self.locked = locked
+
     def get_schema(self):
         class Schema(ma.SQLAlchemySchema):
             class Meta:
@@ -268,6 +273,7 @@ class Share(db.Model):
             owner_name = ma.auto_field()
             file_name = ma.auto_field()
             initialized = ma.auto_field(dump_only=True)
+            locked = ma.auto_field(dump_only=True)
 
         return Schema()
 

tests/conftest.py

@@ -70,6 +70,7 @@ def users(client):
                 Permissions.DELETE,
                 Permissions.MODIFY,
                 Permissions.LIST,
+                Permissions.LOCK,
             ),
         ),
         dave=dict(
@@ -110,6 +111,16 @@ def users(client):
                 Permissions.READ,
             ),
         ),
+        no_lock_user=dict(
+            password="password",
+            permissions=Bitmask(
+                Permissions.CREATE,
+                Permissions.MODIFY,
+                Permissions.DELETE,
+                Permissions.ADMIN,
+                Permissions.READ,
+            ),
+        ),
         administrator=dict(password="4321", permissions=Bitmask(Permissions.ADMIN)),
     )
 

tests/test_files.py

@@ -193,7 +193,6 @@ class TestSuite:
         )
         assert resp.status_code == 403
 
-
         # test not allowing re-upload
         resp = client.post(
             url + "/content",
@@ -210,3 +209,84 @@ class TestSuite:
         assert resp.status_code == 403
         resp = client.get(url + "/content", headers=auth("no_read_user"))
         assert resp.status_code == 403
+
+    def test_locking(self, client, users, auth, rand):
+        # upload share
+        resp = client.post(
+            "/files", headers=auth("jeff"), json={"file_name": "content.bin"}
+        )
+        data = resp.get_json()
+        url = data.get("url")
+        upload_data = rand.randbytes(4000)
+        resp = client.post(
+            url + "/content",
+            headers=auth("jeff"),
+            data={
+                "upload": FileStorage(stream=BytesIO(upload_data), filename="upload")
+            },
+            content_type="multipart/form-data",
+        )
+        assert resp.status_code == 201
+
+        # lock share
+        resp = client.post(
+            url + "/lock",
+            headers=auth("jeff"),
+        )
+        assert resp.status_code == 200
+
+        # attempt to modify share
+        resp = client.put(
+            url + "/content",
+            headers=auth("jeff"),
+            data={
+                "upload": FileStorage(stream=BytesIO(upload_data), filename="upload")
+            },
+            content_type="multipart/form-data",
+        )
+        assert resp.status_code == 423
+
+        # attempt to delete share
+        resp = client.delete(
+            url,
+            headers=auth("jeff"),
+        )
+        assert resp.status_code == 423
+
+
+        # unlock share
+        resp = client.post(
+            url + "/unlock",
+            headers=auth("jeff"),
+        )
+        assert resp.status_code == 200
+
+        # attempt to modify share
+        resp = client.put(
+            url + "/content",
+            headers=auth("jeff"),
+            data={
+                "upload": FileStorage(stream=BytesIO(upload_data), filename="upload")
+            },
+            content_type="multipart/form-data",
+        )
+        assert resp.status_code == 200
+
+        # attempt to delete share
+        resp = client.delete(
+            url,
+            headers=auth("jeff"),
+        )
+        assert resp.status_code == 200
+
+        # attempt to lock/unlock without perms
+        resp = client.post(
+            url + "/lock",
+            headers=auth("no_lock_user"),
+        )
+        assert resp.status_code == 403
+        resp = client.post(
+            url + "/unlock",
+            headers=auth("no_lock_user"),
+        )
+        assert resp.status_code == 403