[DPE-7584] Fix temp tablespace directory permissions (#10)

* Fix temp tablespace directory permissions

Signed-off-by: Marcelo Henrique Neppel <[email protected]>

* Move permissions value to a literal

Signed-off-by: Marcelo Henrique Neppel <[email protected]>

* Move snap user to a literal

Signed-off-by: Marcelo Henrique Neppel <[email protected]>

* Bumb version to 16.0.0

Signed-off-by: Marcelo Henrique Neppel <[email protected]>

---------

Signed-off-by: Marcelo Henrique Neppel <[email protected]>

Author: marceloneppel

pyproject.toml

@@ -4,7 +4,7 @@
 [project]
 name = "postgresql-charms-single-kernel"
 description = "Shared and reusable code for PostgreSQL-related charms"
-version = "0.0.1"
+version = "16.0.0"
 readme = "README.md"
 license = "Apache-2.0"
 authors = [

single_kernel_postgresql/config/literals.py

@@ -5,6 +5,9 @@
 This module should contain the literals used in the charms (paths, enums, etc).
 """
 
+# Permissions.
+POSTGRESQL_STORAGE_PERMISSIONS = 0o700
+
 # Relations.
 PEER = "database-peers"
 
@@ -13,5 +16,6 @@
 MONITORING_USER = "monitoring"
 REPLICATION_USER = "replication"
 REWIND_USER = "rewind"
+SNAP_USER = "_daemon_"
 USER = "operator"
 SYSTEM_USERS = [MONITORING_USER, REPLICATION_USER, REWIND_USER, USER]

single_kernel_postgresql/utils/filesystem.py

@@ -0,0 +1,20 @@
+# Copyright 2025 Canonical Ltd.
+# See LICENSE file for licensing details.
+"""Filesystem utilities."""
+
+import os
+import pwd
+
+from ..config.literals import SNAP_USER
+
+
+def change_owner(path: str) -> None:
+    """Change the ownership of a file or a directory to the snap user.
+
+    Args:
+        path: path to a file or directory.
+    """
+    # Get the uid/gid for the snap user.
+    user_database = pwd.getpwnam(SNAP_USER)
+    # Set the correct ownership for the file or directory.
+    os.chown(path, uid=user_database.pw_uid, gid=user_database.pw_gid)

single_kernel_postgresql/utils/postgresql.py

@@ -20,14 +20,16 @@
 """
 
 import logging
+import os
 from collections import OrderedDict
 from typing import Dict, List, Optional, Set, Tuple
 
 import psycopg2
 from ops import ConfigData
 from psycopg2.sql import SQL, Identifier, Literal
 
-from ..config.literals import BACKUP_USER, SYSTEM_USERS
+from ..config.literals import BACKUP_USER, POSTGRESQL_STORAGE_PERMISSIONS, SYSTEM_USERS
+from .filesystem import change_owner
 
 # Groups to distinguish HBA access
 ACCESS_GROUP_IDENTITY = "identity_access"
@@ -1071,6 +1073,10 @@ def set_up_database(self, temp_location: Optional[str] = None) -> None:
             cursor = connection.cursor()
 
             if temp_location is not None:
+                # Fix permissions on the temporary tablespace location when a reboot happens and tmpfs is being used.
+                change_owner(temp_location)
+                os.chmod(temp_location, POSTGRESQL_STORAGE_PERMISSIONS)
+
                 cursor.execute("SELECT TRUE FROM pg_tablespace WHERE spcname='temp';")
                 if cursor.fetchone() is None:
                     cursor.execute(f"CREATE TABLESPACE temp LOCATION '{temp_location}';")

tests/unit/test_filesystem.py

@@ -0,0 +1,50 @@
+# Copyright 2025 Canonical Ltd.
+# See LICENSE file for licensing details.
+from tempfile import NamedTemporaryFile
+from unittest.mock import MagicMock, patch
+
+import pytest
+from single_kernel_postgresql.utils.filesystem import change_owner
+
+
+def test_change_owner_calls_pwd_and_os_chown_with_daemon_user():
+    with (
+        patch("single_kernel_postgresql.utils.filesystem.pwd.getpwnam") as getpwnam,
+        patch("single_kernel_postgresql.utils.filesystem.os.chown") as chown,
+        NamedTemporaryFile(delete=True) as tmp,
+    ):
+        # Simulate pwd entry
+        pw_entry = MagicMock()
+        pw_entry.pw_uid = 1234
+        pw_entry.pw_gid = 4321
+        getpwnam.return_value = pw_entry
+
+        change_owner(tmp.name)
+
+        getpwnam.assert_called_once_with("_daemon_")
+        chown.assert_called_once_with(tmp.name, uid=1234, gid=4321)
+
+
+def test_change_owner_raises_when_user_missing():
+    # When the _daemon_ user is not present, pwd.getpwnam raises KeyError
+    with (
+        patch("single_kernel_postgresql.utils.filesystem.pwd.getpwnam", side_effect=KeyError),
+        pytest.raises(KeyError),
+        NamedTemporaryFile(delete=True) as tmp,
+    ):
+        change_owner(tmp.name)
+
+
+def test_change_owner_bubbles_up_os_error():
+    # Ensure we surface OSError coming from os.chown
+    with (
+        patch("single_kernel_postgresql.utils.filesystem.pwd.getpwnam") as getpwnam,
+        patch("single_kernel_postgresql.utils.filesystem.os.chown", side_effect=OSError("denied")),
+        NamedTemporaryFile(delete=True) as tmp,
+    ):
+        entry = MagicMock()
+        entry.pw_uid = 1
+        entry.pw_gid = 1
+        getpwnam.return_value = entry
+        with pytest.raises(OSError):
+            change_owner(tmp.name)

tests/unit/test_postgresql.py

@@ -11,9 +11,11 @@
 from single_kernel_postgresql.utils.postgresql import (
     ACCESS_GROUP_INTERNAL,
     ACCESS_GROUPS,
+    ROLE_DATABASES_OWNER,
     PostgreSQL,
     PostgreSQLCreateDatabaseError,
     PostgreSQLCreateUserError,
+    PostgreSQLDatabasesSetupError,
     PostgreSQLGetLastArchivedWALError,
     PostgreSQLUndefinedHostError,
     PostgreSQLUndefinedPasswordError,
@@ -315,6 +317,105 @@ def test_validate_group_map(harness):
         assert harness.charm.postgresql.validate_group_map("ldap_group ldap_test_group") is False
 
 
+def test_set_up_database_with_temp_tablespace_and_missing_owner_role(harness):
+    with (
+        patch(
+            "single_kernel_postgresql.utils.postgresql.PostgreSQL._connect_to_database"
+        ) as _connect_to_database,
+        patch("single_kernel_postgresql.utils.postgresql.PostgreSQL.set_up_login_hook_function"),
+        patch(
+            "single_kernel_postgresql.utils.postgresql.PostgreSQL.set_up_predefined_catalog_roles_function"
+        ),
+        patch("single_kernel_postgresql.utils.postgresql.PostgreSQL.create_user") as _create_user,
+        patch("single_kernel_postgresql.utils.postgresql.change_owner") as _change_owner,
+        patch("single_kernel_postgresql.utils.postgresql.os.chmod") as _chmod,
+    ):
+        # First connection (non-context) for temp tablespace
+        execute_direct = _connect_to_database.return_value.cursor.return_value.execute
+        fetchone_direct = _connect_to_database.return_value.cursor.return_value.fetchone
+        fetchone_direct.return_value = None
+
+        # Second and third connections are context-managed
+        execute_cm = _connect_to_database.return_value.__enter__.return_value.cursor.return_value.__enter__.return_value.execute
+        fetchone_cm = _connect_to_database.return_value.__enter__.return_value.cursor.return_value.__enter__.return_value.fetchone
+        fetchone_cm.return_value = None  # owner role missing
+
+        harness.charm.postgresql.set_up_database(temp_location="/var/lib/postgresql/tmp")
+
+        # Ensure permission fixes applied
+        _change_owner.assert_called_once_with("/var/lib/postgresql/tmp")
+        _chmod.assert_called_once_with("/var/lib/postgresql/tmp", 0o700)
+
+        # Validate temp tablespace operations
+        execute_direct.assert_has_calls([
+            call("SELECT TRUE FROM pg_tablespace WHERE spcname='temp';"),
+            call("CREATE TABLESPACE temp LOCATION '/var/lib/postgresql/tmp';"),
+            call("GRANT CREATE ON TABLESPACE temp TO public;"),
+        ])
+
+        # create_user called for missing owner role
+        _create_user.assert_called_once_with(
+            ROLE_DATABASES_OWNER, can_create_database=True, extra_user_roles=["charmed_dml"]
+        )
+
+        # Final revokes and grants
+        system_users = harness.charm.postgresql.system_users
+        expected = [
+            call("REVOKE ALL PRIVILEGES ON DATABASE postgres FROM PUBLIC;"),
+            call("REVOKE CREATE ON SCHEMA public FROM PUBLIC;"),
+            *[
+                call(SQL("GRANT ALL PRIVILEGES ON DATABASE postgres TO {};").format(Identifier(u)))
+                for u in system_users
+            ],
+        ]
+        execute_cm.assert_has_calls(expected, any_order=False)
+
+
+def test_set_up_database_no_temp_and_existing_owner_role(harness):
+    with (
+        patch(
+            "single_kernel_postgresql.utils.postgresql.PostgreSQL._connect_to_database"
+        ) as _connect_to_database,
+        patch("single_kernel_postgresql.utils.postgresql.PostgreSQL.set_up_login_hook_function"),
+        patch(
+            "single_kernel_postgresql.utils.postgresql.PostgreSQL.set_up_predefined_catalog_roles_function"
+        ),
+        patch("single_kernel_postgresql.utils.postgresql.PostgreSQL.create_user") as _create_user,
+    ):
+        # owner role exists
+        fetchone = _connect_to_database.return_value.__enter__.return_value.cursor.return_value.__enter__.return_value.fetchone
+        fetchone.return_value = True
+
+        harness.charm.postgresql.set_up_database()
+
+        _create_user.assert_not_called()
+
+        execute = _connect_to_database.return_value.__enter__.return_value.cursor.return_value.__enter__.return_value.execute
+        system_users = harness.charm.postgresql.system_users
+        execute.assert_has_calls([
+            call("REVOKE ALL PRIVILEGES ON DATABASE postgres FROM PUBLIC;"),
+            call("REVOKE CREATE ON SCHEMA public FROM PUBLIC;"),
+            *[
+                call(SQL("GRANT ALL PRIVILEGES ON DATABASE postgres TO {};").format(Identifier(u)))
+                for u in system_users
+            ],
+        ])
+
+
+def test_set_up_database_raises_wrapped_error(harness):
+    with (
+        patch(
+            "single_kernel_postgresql.utils.postgresql.PostgreSQL._connect_to_database"
+        ) as _connect_to_database,
+        patch("single_kernel_postgresql.utils.postgresql.change_owner"),
+        patch("single_kernel_postgresql.utils.postgresql.os.chmod"),
+    ):
+        execute_direct = _connect_to_database.return_value.cursor.return_value.execute
+        execute_direct.side_effect = psycopg2.Error
+        with pytest.raises(PostgreSQLDatabasesSetupError):
+            harness.charm.postgresql.set_up_database(temp_location="/tmp")
+
+
 def test_connect_to_database():
     # Error on no host
     pg = PostgreSQL(None, None, "operator", None, "postgres", None)