security(storage): restrict backend credentials to storage administrators

spp_storage_backend/README.rst

@@ -115,6 +115,18 @@ Dependencies
 Changelog
 =========
 
+19.0.2.0.1
+~~~~~~~~~~
+
+- fix(security): restrict storage backend credentials to storage
+  administrators. The S3 access/secret keys and Azure connection string
+  were readable by any internal user via RPC (``base.group_user`` had
+  model read and the fields had no field-level protection). The three
+  credential fields are now gated with
+  ``groups="spp_storage_backend.group_storage_admin"``, and the S3/Azure
+  client builders resolve them via ``sudo()`` so a non-admin can still
+  operate a backend without being able to read the secret.
+
 19.0.2.0.0
 ~~~~~~~~~~
 

spp_storage_backend/__manifest__.py

@@ -4,7 +4,7 @@
     "name": "OpenSPP Storage Backend",
     "summary": "Pluggable storage backend configuration for OpenSPP file storage",
     "category": "OpenSPP/Core",
-    "version": "19.0.2.0.0",
+    "version": "19.0.2.0.1",
     "author": "OpenSPP.org",
     "website": "https://github.com/OpenSPP/OpenSPP2",
     "license": "LGPL-3",

spp_storage_backend/models/storage_backend.py

@@ -58,10 +58,12 @@ class StorageBackend(models.Model):
     )
     s3_access_key = fields.Char(
         string="Access Key",
+        groups="spp_storage_backend.group_storage_admin",
         help="AWS Access Key ID or S3-compatible access key",
     )
     s3_secret_key = fields.Char(
         string="Secret Key",
+        groups="spp_storage_backend.group_storage_admin",
         help="AWS Secret Access Key or S3-compatible secret key",
     )
     s3_region = fields.Char(
@@ -78,6 +80,7 @@ class StorageBackend(models.Model):
     # Azure Configuration
     azure_connection_string = fields.Char(
         string="Connection String",
+        groups="spp_storage_backend.group_storage_admin",
         help="Azure Storage connection string",
     )
     azure_container = fields.Char(
@@ -363,9 +366,14 @@ def _get_s3_client(self):
         except ImportError as e:
             raise UserError(_("boto3 library is not installed. Please install it to use S3 storage.")) from e
 
+        # Credentials are group-restricted (admin-only readable); resolve them
+        # server-side so an operating user can use the backend without being
+        # able to read the secret. The value is only used to build the client.
+        # nosemgrep: odoo-sudo-without-context - read admin-only credential fields to build the client
+        creds = self.sudo()
         config = {
-            "aws_access_key_id": self.s3_access_key,
-            "aws_secret_access_key": self.s3_secret_key,
+            "aws_access_key_id": creds.s3_access_key,
+            "aws_secret_access_key": creds.s3_secret_key,
             "region_name": self.s3_region,
         }
 
@@ -443,6 +451,16 @@ def _test_s3_connection(self):
     # AZURE BACKEND IMPLEMENTATION
     # ========================================================================
 
+    def _get_azure_connection_string(self):
+        """Resolve the group-restricted Azure connection string server-side.
+
+        The field is admin-only readable; an operating user may use the backend
+        without being able to read the secret, so it is read via ``sudo()`` and
+        only used to build clients / SAS URLs (never returned to the caller).
+        """
+        # nosemgrep: odoo-sudo-without-context - read admin-only credential field to build the client
+        return self.sudo().azure_connection_string
+
     def _get_azure_client(self):
         """Get configured Azure Blob client."""
         try:
@@ -452,7 +470,7 @@ def _get_azure_client(self):
                 _("azure-storage-blob library is not installed. Please install it to use Azure storage.")
             ) from e
 
-        service_client = BlobServiceClient.from_connection_string(self.azure_connection_string)
+        service_client = BlobServiceClient.from_connection_string(self._get_azure_connection_string())
         return service_client.get_container_client(self.azure_container)
 
     def _store_azure(self, binary_data, path):
@@ -497,8 +515,10 @@ def _get_azure_public_url(self, reference, expires_in):
         container_client = self._get_azure_client()
         blob_client = container_client.get_blob_client(reference)
 
-        # Extract account name and key from connection string
-        conn_parts = dict(item.split("=", 1) for item in self.azure_connection_string.split(";") if "=" in item)
+        # Extract account name and key from connection string (admin-only field
+        # resolved via sudo; used only to sign the SAS URL, not returned).
+        connection_string = self._get_azure_connection_string()
+        conn_parts = dict(item.split("=", 1) for item in connection_string.split(";") if "=" in item)
         account_name = conn_parts.get("AccountName")
         account_key = conn_parts.get("AccountKey")
 

spp_storage_backend/readme/HISTORY.md

@@ -1,3 +1,13 @@
+### 19.0.2.0.1
+
+- fix(security): restrict storage backend credentials to storage administrators.
+  The S3 access/secret keys and Azure connection string were readable by any
+  internal user via RPC (`base.group_user` had model read and the fields had no
+  field-level protection). The three credential fields are now gated with
+  `groups="spp_storage_backend.group_storage_admin"`, and the S3/Azure client
+  builders resolve them via `sudo()` so a non-admin can still operate a backend
+  without being able to read the secret.
+
 ### 19.0.2.0.0
 
 - Initial migration to OpenSPP2

spp_storage_backend/static/description/index.html

@@ -487,6 +487,19 @@ <h2><a class="toc-backref" href="#toc-entry-1">Changelog</a></h2>
 </div>
 </div>
 <div class="section" id="section-1">
+<h1>19.0.2.0.1</h1>
+<ul class="simple">
+<li>fix(security): restrict storage backend credentials to storage
+administrators. The S3 access/secret keys and Azure connection string
+were readable by any internal user via RPC (<tt class="docutils literal">base.group_user</tt> had
+model read and the fields had no field-level protection). The three
+credential fields are now gated with
+<tt class="docutils literal"><span class="pre">groups=&quot;spp_storage_backend.group_storage_admin&quot;</span></tt>, and the S3/Azure
+client builders resolve them via <tt class="docutils literal">sudo()</tt> so a non-admin can still
+operate a backend without being able to read the secret.</li>
+</ul>
+</div>
+<div class="section" id="section-2">
 <h1>19.0.2.0.0</h1>
 <ul class="simple">
 <li>Initial migration to OpenSPP2</li>

spp_storage_backend/tests/__init__.py

@@ -1,3 +1,4 @@
 # Part of OpenSPP. See LICENSE file for full copyright and licensing details.
 
 from . import test_storage_backend
+from . import test_credential_access

spp_storage_backend/tests/test_credential_access.py

@@ -0,0 +1,159 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+"""Storage backend credentials must not be readable by ordinary internal users.
+
+The model stores S3 access/secret keys and Azure connection strings. Ordinary
+`base.group_user` users get model read access (to resolve/use a backend), but
+must never be able to read the raw credentials over RPC/`read`/`search_read`.
+Only storage administrators (and trusted server-side `sudo()` flows, e.g. the
+S3/Azure client builders) may resolve the credential values.
+"""
+
+import sys
+from unittest.mock import MagicMock, patch
+
+from odoo.exceptions import AccessError
+from odoo.tests.common import TransactionCase, tagged
+
+S3_SECRET = "super-secret-s3-key"
+S3_ACCESS = "super-access-s3-id"
+AZURE_CONN = "DefaultEndpointsProtocol=https;AccountName=x;AccountKey=SECRETKEY==;"
+CRED_FIELDS = ("s3_access_key", "s3_secret_key", "azure_connection_string")
+
+
+@tagged("post_install", "-at_install")
+class TestStorageCredentialAccess(TransactionCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        Backend = cls.env["spp.storage.backend"]
+        cls.s3_backend = Backend.create(
+            {
+                "name": "S3 Cred Backend",
+                "backend_type": "s3",
+                "s3_bucket": "bucket",
+                "s3_access_key": S3_ACCESS,
+                "s3_secret_key": S3_SECRET,
+                "s3_region": "us-east-1",
+            }
+        )
+        cls.azure_backend = Backend.create(
+            {
+                "name": "Azure Cred Backend",
+                "backend_type": "azure",
+                "azure_connection_string": AZURE_CONN,
+                "azure_container": "container",
+            }
+        )
+        cls.base_user = cls.env["res.users"].create(
+            {
+                "name": "Plain Internal User",
+                "login": "storage_plain_user",
+                "group_ids": [(6, 0, [cls.env.ref("base.group_user").id])],
+            }
+        )
+        cls.storage_admin = cls.env["res.users"].create(
+            {
+                "name": "Storage Admin User",
+                "login": "storage_admin_user",
+                "group_ids": [
+                    (
+                        6,
+                        0,
+                        [
+                            cls.env.ref("base.group_user").id,
+                            cls.env.ref("spp_storage_backend.group_storage_admin").id,
+                        ],
+                    )
+                ],
+            }
+        )
+
+    def _read_field(self, backend, user, field):
+        """Return the field value as seen by `user`, or None if blocked."""
+        try:
+            return backend.with_user(user).read([field])[0].get(field)
+        except AccessError:
+            return None
+
+    # --- credentials hidden from ordinary users -----------------------
+
+    def test_base_user_cannot_read_s3_secret(self):
+        self.assertNotEqual(
+            self._read_field(self.s3_backend, self.base_user, "s3_secret_key"),
+            S3_SECRET,
+            "base.group_user must not be able to read s3_secret_key",
+        )
+
+    def test_base_user_cannot_read_s3_access_key(self):
+        self.assertNotEqual(
+            self._read_field(self.s3_backend, self.base_user, "s3_access_key"),
+            S3_ACCESS,
+        )
+
+    def test_base_user_cannot_read_azure_connection_string(self):
+        self.assertNotEqual(
+            self._read_field(self.azure_backend, self.base_user, "azure_connection_string"),
+            AZURE_CONN,
+        )
+
+    def test_base_user_search_read_explicit_credentials_denied(self):
+        # Explicitly requesting a restricted field over RPC raises AccessError.
+        with self.assertRaises(AccessError):
+            self.env["spp.storage.backend"].with_user(self.base_user).search_read([], list(CRED_FIELDS))
+
+    def test_base_user_default_read_excludes_credentials(self):
+        # Reading the accessible fields must not surface the credentials.
+        rows = self.env["spp.storage.backend"].with_user(self.base_user).search_read([])
+        self.assertTrue(rows)
+        for row in rows:
+            for field in CRED_FIELDS:
+                self.assertNotIn(field, row, f"{field} leaked via default search_read")
+
+    def test_credential_fields_are_group_restricted(self):
+        info = self.env["spp.storage.backend"].fields_get(list(CRED_FIELDS))
+        for field in CRED_FIELDS:
+            self.assertTrue(
+                info[field].get("groups"),
+                f"{field} must carry a field-level groups restriction",
+            )
+
+    # --- admins and trusted flows can still resolve them --------------
+
+    def test_storage_admin_can_read_credentials(self):
+        self.assertEqual(
+            self._read_field(self.s3_backend, self.storage_admin, "s3_secret_key"),
+            S3_SECRET,
+        )
+        self.assertEqual(
+            self._read_field(self.azure_backend, self.storage_admin, "azure_connection_string"),
+            AZURE_CONN,
+        )
+
+    def test_s3_client_builder_resolves_credentials_for_non_admin(self):
+        """The S3 client is built server-side with the real credentials even
+        when the operating user cannot read them (use-without-see via sudo)."""
+        fake_boto3 = MagicMock()
+        with patch.dict(sys.modules, {"boto3": fake_boto3}):
+            self.s3_backend.with_user(self.base_user)._get_s3_client()
+        _, kwargs = fake_boto3.client.call_args
+        self.assertEqual(kwargs.get("aws_secret_access_key"), S3_SECRET)
+        self.assertEqual(kwargs.get("aws_access_key_id"), S3_ACCESS)
+
+    def test_azure_client_builder_resolves_credentials_for_non_admin(self):
+        fake_module = MagicMock()
+        mods = {"azure": MagicMock(), "azure.storage": MagicMock(), "azure.storage.blob": fake_module}
+        with patch.dict(sys.modules, mods):
+            self.azure_backend.with_user(self.base_user)._get_azure_client()
+        args, _ = fake_module.BlobServiceClient.from_connection_string.call_args
+        self.assertEqual(args[0], AZURE_CONN)
+
+    def test_azure_public_url_resolves_credentials_for_non_admin(self):
+        """SAS-URL generation reads the connection string through the same sudo
+        helper, so it works for an operating non-admin user (regression: the
+        second Azure credential read site)."""
+        fake_module = MagicMock()
+        mods = {"azure": MagicMock(), "azure.storage": MagicMock(), "azure.storage.blob": fake_module}
+        with patch.dict(sys.modules, mods):
+            self.azure_backend.with_user(self.base_user)._get_azure_public_url("blob.txt", 3600)
+        _, kwargs = fake_module.generate_blob_sas.call_args
+        self.assertEqual(kwargs.get("account_key"), "SECRETKEY==")