Chore: Make data encrytion for services optional (#1918)

* Removing the Singleton

* Updated diagram

* fix tests

* removing simplejsons references

* adding tests for encrypted Storages

* working that only the encrypted data is removed

# Conflicts:
#	src/cryptoadvance/specter/managers/service_manager.py
#	src/cryptoadvance/specter/services/service.py
#	src/cryptoadvance/specter/services/swan/service.py
#	tests/test_services.py

* remove classmethod

* Revert "remove classmethod"

This reverts commit 7b1859d.

* better unlink text

* adding check that the correct services are removed from the users

Co-authored-by: Kim Neunert <[email protected]>

Author: relativisticelectron

src/cryptoadvance/specter/managers/service_manager.py

@@ -328,22 +328,48 @@ def get_service(self, plugin_id: str) -> Service:
             raise ExtensionException(f"No such plugin: '{plugin_id}'")
         return self._services[plugin_id]
 
-    def remove_all_services_from_user(self, user: User):
-        """
-        Clears User.services and `user_secret`; wipes the User's
-        ServiceEncryptedStorage.
-        """
-        # Don't show any Services on the sidebar for the admin user
-        user.services.clear()
+    def delete_service_from_user(self, user: User, service_id: str, autosave=True):
+        "Removes the service for the user and deletes the stored data in the ServiceEncryptedStorage"
+        # remove the service from the sidebar
+        user.remove_service(service_id, autosave=autosave)
+        # delete the data if it was encrypted
+        if (
+            self.user_has_encrypted_storage(user=user)
+            and self.get_service(service_id).encrypt_data
+        ):
+            self.specter.service_encrypted_storage_manager.remove_service_data(
+                user, service_id, autosave=autosave
+            )
+        # here we do not need to delete the data if it was unencrypted
 
-        # Reset as if we never had any encrypted storage
-        user.delete_user_secret(autosave=False)
-        user.save_info()
+    def delete_services_with_encrypted_storage(self, user: User):
+        services_with_encrypted_storage = [
+            service_id
+            for service_id in self.services
+            if self.get_service(service_id).encrypt_data
+        ]
+        for service_id in services_with_encrypted_storage:
+            self.delete_service_from_user(user, service_id, autosave=True)
+
+        user.delete_user_secret(autosave=True)
+        # Encrypted Service data is now orphaned since there is no
+        # password. So wipe it from the disk.
+        self.specter.service_encrypted_storage_manager.delete_all_service_data(user)
+        logger.debug(
+            f"Deleted encrypted services {services_with_encrypted_storage} and user secret"
+        )
+
+    def delete_services_with_unencrypted_storage(self, user: User):
+        services_with_unencrypted_storage = [
+            service_id
+            for service_id in self.services
+            if not self.get_service(service_id).encrypt_data
+        ]
+        for service_id in services_with_unencrypted_storage:
+            self.delete_service_from_user(user, service_id, autosave=True)
 
-        if self.user_has_encrypted_storage(user=user):
-            # Encrypted Service data is now orphaned since there is no
-            # password. So wipe it from the disk.
-            app.specter.service_encrypted_storage_manager.delete_all_service_data(user)
+        self.specter.service_unencrypted_storage_manager.delete_all_service_data(user)
+        logger.debug(f"Deleted unencrypted services")
 
     @classmethod
     def get_service_x_dirs(cls, x):

src/cryptoadvance/specter/server_endpoints/settings.py

@@ -570,12 +570,11 @@ def auth():
                     users = None
                     app.config["LOGIN_DISABLED"] = True
 
-                    # Cannot support Services if there's no password (admin was already
-                    # warned about this in the UI). Remove User.services, clear the
-                    # `user_secret`, and wipe the ServiceEncryptedStorage.
-                    app.specter.service_manager.remove_all_services_from_user(
-                        current_user
-                    )
+                    # if there is no password, we have to delete the previously encrypted data from services
+                    for user in app.specter.user_manager.users:
+                        app.specter.service_manager.delete_services_with_encrypted_storage(
+                            user
+                        )
 
             # Redirect if a URL was given via the next variable
             if request.form.get("next") and request.form.get("next") != "":

src/cryptoadvance/specter/services/service.py

@@ -38,6 +38,7 @@ class Service:
     # If the blueprint gets a "/ext" prefix (isolated_client = True), the login cookie won't work for all specter core functionality
     isolated_client = True
     devstatus = devstatus_alpha
+    encrypt_data = False
 
     def __init__(self, active, specter):
         if not hasattr(self, "id"):
@@ -47,30 +48,34 @@ def __init__(self, active, specter):
         self.active = active
         self.specter = specter
 
+    @classmethod
+    def _storage_manager(cls):
+        return (
+            app.specter.service_encrypted_storage_manager
+            if cls.encrypt_data
+            else app.specter.service_unencrypted_storage_manager
+        )
+
     def callback(self, callback_id, *argv, **kwargv):
         if callback_id == callbacks.after_serverpy_init_app:
             if hasattr(self, "callback_after_serverpy_init_app"):
                 self.callback_after_serverpy_init_app(kwargv["scheduler"])
 
     @classmethod
     def set_current_user_service_data(cls, service_data: dict):
-        app.specter.service_encrypted_storage_manager.set_current_user_service_data(
+        cls._storage_manager().set_current_user_service_data(
             service_id=cls.id, service_data=service_data
         )
 
     @classmethod
     def update_current_user_service_data(cls, service_data: dict):
-        app.specter.service_encrypted_storage_manager.update_current_user_service_data(
+        cls._storage_manager().update_current_user_service_data(
             service_id=cls.id, service_data=service_data
         )
 
     @classmethod
     def get_current_user_service_data(cls) -> dict:
-        return (
-            app.specter.service_encrypted_storage_manager.get_current_user_service_data(
-                service_id=cls.id
-            )
-        )
+        return cls._storage_manager().get_current_user_service_data(service_id=cls.id)
 
     @classmethod
     def get_blueprint_name(cls):

src/cryptoadvance/specter/services/service_encrypted_storage.py

@@ -98,6 +98,16 @@ def get_service_data(self, service_id: str) -> dict:
             service_data = {}
         return service_data
 
+    def remove_service_data(self, service_id: str, autosave: bool = True):
+        # Add or update fields; does not remove existing fields
+        if service_id not in self.data:
+            logger.warning(f"service_id {service_id} does not exist in self.data")
+            return
+
+        del self.data[service_id]
+        if autosave:
+            self._save()
+
 
 class ServiceUnencryptedStorage(ServiceEncryptedStorage):
     """In order to use ServiceEncryptedStorage but unencrypted, we derive from that class
@@ -178,6 +188,17 @@ def delete_all_service_data(self, user: User):
         encrypted_storage.data = {}
         encrypted_storage._save()
 
+    def remove_service_data(self, user: User, service_id: str, autosave: bool = True):
+        if user.id in self.storage_by_user:
+            self.storage_by_user[user].remove_service_data(
+                service_id, autosave=autosave
+            )
+            logger.debug(f"Removed dervice_data from user {user.id}")
+        else:
+            logger.debug(
+                f"Could not remove dervice_data from user {user.id}, because {user.id} was not found in storage_by_user"
+            )
+
 
 class ServiceUnencryptedStorageManager(ServiceEncryptedStorageManager):
     def __init__(self, data_folder, user_manager: UserManager):

src/cryptoadvance/specter/services/swan/service.py

@@ -28,6 +28,7 @@ class SwanService(Service):
     has_blueprint = True
     isolated_client = False
     devstatus = devstatus_prod
+    encrypt_data = True
 
     # TODO: As more Services are integrated, we'll want more robust categorization and sorting logic
     sort_priority = 1

src/cryptoadvance/specter/templates/settings/auth_settings.jinja

@@ -28,7 +28,9 @@
 				<br><br>
 			{% endif %}
 			<div id="hasencryptedservicedata" style="display: none" class="hasencryptedservicedata note">
-				<p>{{ _("Note: If you set Authentication to \"None\", Specter will unlink your Service integrations as a security precaution.") }}</p>
+				<p>{{ _("Note: If you set Authentication to \"None\", Specter will unlink the Service integrations ") }}  
+					"{{ specter.service_manager.services_sorted | selectattr('encrypt_data') | map(attribute='name') | join('", "') }}"
+					{{ _(" as a security precaution.") }}</p>
 			</div>
 			<div id="ratelimit" class="{% if method == 'none' or not current_user.is_admin %}hidden{% endif %}">
 				{{ _("Rate Limiting (seconds between login/register attempts)") }}:<br><input id="rate_limit" type="number" name="rate_limit" min="0" max="3600" step="1" value="{{ rate_limit }}" required><br><br>

tests/test_services.py

@@ -23,11 +23,20 @@
 from cryptoadvance.specter.user import User, hash_password
 
 
+class FakeServiceNoEncryption(Service):
+    # A dummy Service just used by the test suite
+    id = "test_service_no_encryption"
+    name = "Test Service no encryption"
+    has_blueprint = False
+    encrypt_data = False
+
+
 class FakeService(Service):
     # A dummy Service just used by the test suite
     id = "test_service"
     name = "Test Service"
     has_blueprint = False
+    encrypt_data = True
 
 
 # @patch("cryptoadvance.specter.services.service_manager.app")
@@ -158,7 +167,9 @@ def test_access_encrypted_storage_after_login(app_no_node: SpecterFlask):
         ) == {"somekey": "green"}
 
 
-def test_remove_all_services_from_user(app_no_node: SpecterFlask, empty_data_folder):
+def test_remove_encrypted_services_from_user(
+    app_no_node: SpecterFlask, empty_data_folder
+):
     """ServiceEncryptedStorage should be accessible (decryptable) after user login"""
     # Create test users; automatically generates their `user_secret` and kept decrypted
     # in memory.
@@ -170,9 +181,8 @@ def test_remove_all_services_from_user(app_no_node: SpecterFlask, empty_data_fol
         config={},
     )
 
-    storage_manager = ServiceEncryptedStorageManager(
-        user_manager.data_folder, user_manager
-    )
+    storage_manager = app_no_node.specter.service_encrypted_storage_manager
+    service_manager = app_no_node.specter.service_manager
     storage_manager.storage_by_user = {}
 
     # Need a simulated request context to enable `current_user` lookup
@@ -199,8 +209,22 @@ def test_remove_all_services_from_user(app_no_node: SpecterFlask, empty_data_fol
         # Can't test the actual values because they're encrypted, but the Service.id key is plaintext
         assert FakeService.id in data_on_disk
 
-        # Now remove all
-        app_no_node.specter.service_manager.remove_all_services_from_user(user)
+        # Remove all services that need encryption
+        # we add the fakeservice to the service_manager.services otherwise delete_services_with_encrypted_storage doesn't know it exists
+        # strictly speaking the important call is here user.delete_user_secret(autosave=True) which will execute regardless of adding fakeservice
+        fake_service = FakeService(True, app_no_node.specter)
+        service_manager.services[fake_service.id] = fake_service
+        assert fake_service.id in service_manager.services
+
+        # also add it to the user, and check later it was remove from the user
+        user.add_service(fake_service.id)
+        assert user.has_service(fake_service.id)
+
+        app_no_node.specter.service_manager.delete_services_with_encrypted_storage(user)
+        # the user should not have the fake_service activated any more
+        assert not user.has_service(fake_service.id)
+        # the service_manager on the other hand keeps all services, no matter what
+        assert service_manager.services[fake_service.id]
 
         # Verify data on disk; Bob's user should have his user_secret cleared.
         users_file = app_no_node.specter.user_manager.users_file
@@ -230,6 +254,69 @@ def test_remove_all_services_from_user(app_no_node: SpecterFlask, empty_data_fol
         assert data_on_disk == {}
 
 
+def test_check_differences_between_encrypted_and_non_encrypted_services(
+    app_no_node: SpecterFlask, empty_data_folder
+):
+    """ServiceEncryptedStorage should be accessible (decryptable) after user login"""
+    # Create test users; automatically generates their `user_secret` and kept decrypted
+    # in memory.
+    user_manager: UserManager = app_no_node.specter.user_manager
+    user_manager.create_user(
+        user_id="bob",
+        username="bob",
+        plaintext_password="plain_pass_bob",
+        config={},
+    )
+
+    service_manager = app_no_node.specter.service_manager
+    user = user_manager.get_user("bob")
+
+    def setup_services():
+        # Remove all services that need encryption
+        # we add the fakeservice to the service_manager.services otherwise delete_services_with_encrypted_storage doesn't know it exists
+        # strictly speaking the important call is here user.delete_user_secret(autosave=True) which will execute regardless of adding fakeservice
+        fake_service = FakeService(True, app_no_node.specter)
+        fake_service_no_encryption = FakeServiceNoEncryption(True, app_no_node.specter)
+        service_manager.services[fake_service.id] = fake_service
+        service_manager.services[
+            fake_service_no_encryption.id
+        ] = fake_service_no_encryption
+        assert fake_service.id in service_manager.services
+        assert fake_service_no_encryption.id in service_manager.services
+
+        # also add it to the user, and check later it was remove from the user
+        user.add_service(fake_service.id)
+        user.add_service(fake_service_no_encryption.id)
+        assert user.has_service(fake_service.id)
+        assert user.has_service(fake_service_no_encryption.id)
+
+        return fake_service, fake_service_no_encryption
+
+    fake_service, fake_service_no_encryption = setup_services()
+    # delete the encrypted ones
+    app_no_node.specter.service_manager.delete_services_with_encrypted_storage(user)
+    assert not user.has_service(fake_service.id)
+    assert user.has_service(fake_service_no_encryption.id)
+    # delete the unencrypted ones
+    app_no_node.specter.service_manager.delete_services_with_unencrypted_storage(user)
+    assert not user.has_service(fake_service_no_encryption.id)
+
+    # now setup again and check a different order of execution
+    fake_service, fake_service_no_encryption = setup_services()
+    # delete the unencrypted ones
+    app_no_node.specter.service_manager.delete_services_with_unencrypted_storage(user)
+    assert not user.has_service(fake_service_no_encryption.id)
+    assert user.has_service(fake_service.id)
+    # delete the encrypted ones
+    app_no_node.specter.service_manager.delete_services_with_encrypted_storage(user)
+    # the user should not have the fake_service activated any more
+    assert not user.has_service(fake_service.id)
+
+    # the service_manager on the other hand keeps all services, no matter what
+    assert service_manager.services[fake_service.id]
+    assert service_manager.services[fake_service_no_encryption.id]
+
+
 def test_ServiceUnEncryptedStorage(empty_data_folder, user1, user2):
     user1._generate_user_secret("muh")