Delete tail files

aries_cloudagent/revocation/routes.py

@@ -2,6 +2,8 @@
 
 import json
 import logging
+import os
+import shutil
 from asyncio import shield
 import re
 
@@ -146,6 +148,31 @@ def validate_fields(self, data, **kwargs):
     )
 
 
+class RevRegId(OpenAPISchema):
+    """Parameters and validators for delete tails file request."""
+
+    @validates_schema
+    def validate_fields(self, data, **kwargs):
+        """Validate schema fields - must have either rr-id or cr-id."""
+
+        rev_reg_id = data.get("rev_reg_id")
+        cred_def_id = data.get("cred_def_id")
+
+        if not (rev_reg_id or cred_def_id):
+            raise ValidationError("Request must have either rev_reg_id or cred_def_id")
+
+    rev_reg_id = fields.Str(
+        description="Revocation registry identifier",
+        required=False,
+        **INDY_REV_REG_ID,
+    )
+    cred_def_id = fields.Str(
+        description="Credential definition identifier",
+        required=False,
+        **INDY_CRED_DEF_ID,
+    )
+
+
 class RevokeRequestSchema(CredRevRecordQueryStringSchema):
     """Parameters and validators for revocation request."""
 
@@ -1467,6 +1494,51 @@ async def on_revocation_registry_endorsed_event(profile: Profile, event: Event):
         )
 
 
+@querystring_schema(RevRegId())
+@docs(tags=["revocation"], summary="Delete the tail files")
+async def delete_tails(request: web.BaseRequest) -> json:
+    """Delete Tails Files."""
+    context: AdminRequestContext = request["context"]
+    rev_reg_id = request.query.get("rev_reg_id")
+    cred_def_id = request.query.get("cred_def_id")
+    revoc = IndyRevocation(context.profile)
+    session = revoc._profile.session()
+    if rev_reg_id:
+        rev_reg = await revoc.get_issuer_rev_reg_record(rev_reg_id)
+        tails_path = rev_reg.tails_local_path
+        main_dir_rev = os.path.dirname(tails_path)
+        try:
+            shutil.rmtree(main_dir_rev)
+            return web.json_response({"message": "All files deleted successfully"})
+        except Exception as e:
+            return web.json_response({"message": str(e)})
+    elif cred_def_id:
+        async with session:
+            cred_reg = sorted(
+                await IssuerRevRegRecord.query_by_cred_def_id(
+                    session, cred_def_id, IssuerRevRegRecord.STATE_GENERATED
+                )
+            )[0]
+        tails_path = cred_reg.tails_local_path
+        main_dir_rev = os.path.dirname(tails_path)
+        main_dir_cred = os.path.dirname(main_dir_rev)
+        filenames = os.listdir(main_dir_cred)
+        try:
+            flag = 0
+            for i in filenames:
+                safe_cred_def_id = re.escape(cred_def_id)
+                if re.search(safe_cred_def_id, i):
+                    shutil.rmtree(main_dir_cred + "/" + i)
+                    flag = 1
+            if flag:
+                return web.json_response({"message": "All files deleted successfully"})
+            else:
+                return web.json_response({"message": "No such file or directory"})
+
+        except Exception as e:
+            return web.json_response({"message": str(e)})
+
+
 async def register(app: web.Application):
     """Register routes."""
     app.add_routes(
@@ -1524,6 +1596,7 @@ async def register(app: web.Application):
                 "/revocation/registry/{rev_reg_id}/fix-revocation-entry-state",
                 update_rev_reg_revoked_state,
             ),
+            web.delete("/revocation/registry/delete-tails-file", delete_tails),
         ]
     )
 

aries_cloudagent/revocation/tests/test_routes.py

@@ -1,3 +1,7 @@
+import os
+import shutil
+import unittest
+
 from aiohttp.web import HTTPBadRequest, HTTPNotFound
 from asynctest import TestCase as AsyncTestCase
 from asynctest import mock as async_mock
@@ -903,3 +907,63 @@ async def test_post_process_routes(self):
         ]["get"]["responses"]["200"]["schema"] == {"type": "string", "format": "binary"}
 
         assert "tags" in mock_app._state["swagger_dict"]
+
+
+class TestDeleteTails(unittest.TestCase):
+    def setUp(self):
+        self.rev_reg_id = "rev_reg_id_123"
+        self.cred_def_id = "cred_def_id_456"
+
+        self.main_dir_rev = "path/to/main/dir/rev"
+        self.tails_path = os.path.join(self.main_dir_rev, "tails")
+        if not (os.path.exists(self.main_dir_rev)):
+            os.makedirs(self.main_dir_rev)
+        open(self.tails_path, "w").close()
+
+    async def test_delete_tails_by_rev_reg_id(self):
+        # Setup
+        rev_reg_id = self.rev_reg_id
+
+        # Test
+        result = await test_module.delete_tails(
+            {"context": None, "query": {"rev_reg_id": rev_reg_id}}
+        )
+
+        # Assert
+        self.assertEqual(result, {"message": "All files deleted successfully"})
+        self.assertFalse(os.path.exists(self.tails_path))
+
+    async def test_delete_tails_by_cred_def_id(self):
+        # Setup
+        cred_def_id = self.cred_def_id
+        main_dir_cred = "path/to/main/dir/cred"
+        os.makedirs(main_dir_cred)
+        cred_dir = os.path.join(main_dir_cred, cred_def_id)
+        os.makedirs(cred_dir)
+
+        # Test
+        result = await test_module.delete_tails(
+            {"context": None, "query": {"cred_def_id": cred_def_id}}
+        )
+
+        # Assert
+        self.assertEqual(result, {"message": "All files deleted successfully"})
+        self.assertFalse(os.path.exists(cred_dir))
+        self.assertTrue(os.path.exists(main_dir_cred))
+
+    async def test_delete_tails_not_found(self):
+        # Setup
+        cred_def_id = "invalid_cred_def_id"
+
+        # Test
+        result = await test_module.delete_tails(
+            {"context": None, "query": {"cred_def_id": cred_def_id}}
+        )
+
+        # Assert
+        self.assertEqual(result, {"message": "No such file or directory"})
+        self.assertTrue(os.path.exists(self.main_dir_rev))
+
+    async def tearDown(self):
+        if os.path.exists(self.main_dir_rev):
+            shutil.rmtree(self.main_dir_rev)