Repository: zalando/patroni (python)

Issue:
allow-group-access clause in initdb seems not woking correctly
**Describe the bug**
When using allow-group-access clause in initdb still have some files stlil created with 600 permisstions
```
find /data/postgresql/11/data/ -type f  -perm 600
/data/postgresql/11/data/patroni.dynamic.json
/data/postgresql/11/data/postgresql.base.conf.backup
/data/postgresql/11/data/postgresql.conf.backup
/data/postgresql/11/data/recovery.done
```
As far as I know first 3 files are definitely created by patroni. recovery.* file also appears after restarting patroni daemon.

**To Reproduce**
1. Create patroni config with allow-group-access option
```
initdb:
    - allow-group-access
```
2. (Re)initialize cluster.
3. Observe file permissions in data directory.

**Expected behavior**
All files mast have group access permissions

**Environment**
- Patroni version: patroni 1.6.5
- PostgreSQL version: PostgreSQL 11.8 on x86_64-pc-linux-gnu, compiled by gcc (Ubuntu 7.5.0-3ubuntu1~18.04) 7.5.0, 64-bit
- DCS (and its version): etcd Version: 3.2.17

I am not allowed to upgrade PostgreSQL and Patroni versions right now.

**Patroni configuration file**
Not sure it's needed. Please let me know if it is.

**patronictl show-config**
```
patronictl -c /etc/patroni/config.yml show-config
loop_wait: 10
maximum_lag_on_failover: 1048576
postgresql:
  parameters:
    max_connections: '200'
    max_locks_per_transaction: '64'
    max_prepared_transactions: '3'
    max_worker_processes: '1'
    track_commit_timestamp: false
    wal_level: replica
    wal_log_hints: true
  use_pg_rewind: true
  use_slots: true
retry_timeout: 10
ttl: 30
```

**Have you checked Patroni logs?**
Not sure it's needed. Please let me know if it is.

**Have you checked PostgreSQL logs?**
Not sure it's needed. Please let me know if it is.

**Have you tried to use GitHub issue search?**
Was unable to find anything usefull for allow-group-access


Pull request description:
Set permissions for files and directories created in PGDATA
Postgres supports two types of permissions:
1. owner only
2. group readable

By default the first one is used because it provides better security. But, sometimes people want to run a backup tool with the user that is different from postgres. In this case the second option becomes very useful. Unfortunately it didn't work correctly because Patroni was creating files with owner access only permissions.

This PR changes the behaviour and permissions on files and directories that are created by Patroni will be calculated based on permissions of PGDATA. I.e., they will get group readable access when it is necessary.

Close #1899
Close #1901

Patch:
diff --git a/docs/dynamic_configuration.rst b/docs/dynamic_configuration.rst
index 5486bf6e..f5bb4ee5 100644
--- a/docs/dynamic_configuration.rst
+++ b/docs/dynamic_configuration.rst
@@ -46,9 +46,9 @@ In order to change the dynamic configuration you can use either ``patronictl edi
    -  **archive\_cleanup\_command**: cleanup command for standby leader
    -  **recovery\_min\_apply\_delay**: how long to wait before actually apply WAL records on a standby leader
 
--  **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent logical replication slots requires **postgresql.use_slots** to be set and will also automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+.
+-  **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Permanent slots that don't exist will be created by Patroni. The physical slots are maintained only in the current primary. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent logical replication slots requires **postgresql.use_slots** to be set and will also automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+.
 
-   -  **my\_slot\_name**: the name of replication slot. If the permanent slot name matches with the name of the current primary it will not be created. Everything else is the responsibility of the operator to make sure that there are no clashes in names between replication slots automatically created by Patroni for members and permanent replication slots.
+   -  **my\_slot\_name**: the name of the permanent replication slot. If the permanent slot name matches with the name of the current leader it will not be created. Please note that Patroni does not make checks for permanent slot names added to this configuration matching those that Patroni creates automatically for members. If those names are added, Patroni will ensure that any slots that were created are not removed even if the member becomes unresponsive, situation which would normally result in the slot's removal by Patroni. Although this can be useful in some situations, such as when importing existing members to a new Patroni cluster (see :ref:`Convert a Standalone to a Patroni Cluster <existing_data>` for details), caution should be exercised by the operator that these clashes in names are not persisted in the DCS due to its effect on normal functioning of Patroni.
 
       -  **type**: slot type. Could be ``physical`` or ``logical``. If the slot is logical, you have to additionally define ``database`` and ``plugin``.
       -  **database**: the database name where logical slots should be created.
diff --git a/patroni/config.py b/patroni/config.py
index facf16fd..d2770517 100644
--- a/patroni/config.py
+++ b/patroni/config.py
@@ -13,6 +13,7 @@ from . import PATRONI_ENV_PREFIX
 from .collections import CaseInsensitiveDict
 from .dcs import ClusterConfig, Cluster
 from .exceptions import ConfigParseError
+from .file_perm import pg_perm
 from .postgresql.config import ConfigHandler
 from .utils import deep_compare, parse_bool, parse_int, patch_config
 
@@ -275,11 +276,13 @@ class Config(object):
         if self._cache_needs_saving:
             tmpfile = fd = None
             try:
+                pg_perm.set_permissions_from_data_directory(self._data_dir)
                 (fd, tmpfile) = tempfile.mkstemp(prefix=self.__CACHE_FILENAME, dir=self._data_dir)
                 with os.fdopen(fd, 'w') as f:
                     fd = None
                     json.dump(self.dynamic_configuration, f)
                 tmpfile = shutil.move(tmpfile, self._cache_file)
+                os.chmod(self._cache_file, pg_perm.file_create_mode)
                 self._cache_needs_saving = False
             except Exception:
                 logger.exception('Exception when saving file: %s', self._cache_file)
diff --git a/patroni/ctl.py b/patroni/ctl.py
index 312c0cec..b1d63d2b 100644
--- a/patroni/ctl.py
+++ b/patroni/ctl.py
@@ -36,7 +36,7 @@ from collections import defaultdict
 from contextlib import contextmanager
 from prettytable import ALL, FRAME, PrettyTable
 from urllib.parse import urlparse
-from typing import Any, Dict, Generator, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING
+from typing import Any, Dict, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING
 if TYPE_CHECKING:  # pragma: no cover
     from psycopg import Cursor
     from psycopg2 import cursor
@@ -1816,7 +1816,7 @@ def resume(obj: Dict[str, Any], cluster_name: str, group: Optional[int], wait: b
 
 
 @contextmanager
-def temporary_file(contents: bytes, suffix: str = '', prefix: str = 'tmp') -> Generator[str, None, None]:
+def temporary_file(contents: bytes, suffix: str = '', prefix: str = 'tmp') -> Iterator[str]:
     """Create a temporary file with specified contents that persists for the context.
 
     :param contents: binary string that will be written to the file.
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index 29a4d766..4bcac3ce 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -555,36 +555,50 @@ class TimelineHistory(NamedTuple):
         return TimelineHistory(version, value, lines)
 
 
-class Cluster(NamedTuple):
-    """Immutable object (namedtuple) which represents PostgreSQL cluster.
+class Cluster(NamedTuple('Cluster',
+                         [('initialize', Optional[str]),
+                          ('config', Optional[ClusterConfig]),
+                          ('leader', Optional[Leader]),
+                          ('last_lsn', int),
+                          ('members', List[Member]),
+                          ('failover', Optional[Failover]),
+                          ('sync', SyncState),
+                          ('history', Optional[TimelineHistory]),
+                          ('slots', Optional[Dict[str, int]]),
+                          ('failsafe', Optional[Dict[str, str]]),
+                          ('workers', Dict[int, 'Cluster'])])):
+    """Immutable object (namedtuple) which represents PostgreSQL or Citus cluster.
+
+    .. note::
+        We are using an old-style attribute declaration here because otherwise it is not possible to override `__new__`
+        method. Without it the *workers* by default gets always the same :class:`dict` object that could be mutated.
+
     Consists of the following fields:
-    :param initialize: shows whether this cluster has initialization key stored in DC or not.
-    :param config: global dynamic configuration, reference to `ClusterConfig` object
-    :param leader: `Leader` object which represents current leader of the cluster
-    :param last_lsn: int or long object containing position of last known leader LSN.
-        This value is stored in the `/status` key or `/optime/leader` (legacy) key
-    :param members: list of Member object, all PostgreSQL cluster members including leader
-    :param failover: reference to `Failover` object
-    :param sync: reference to `SyncState` object, last observed synchronous replication state.
-    :param history: reference to `TimelineHistory` object
-    :param slots: state of permanent logical replication slots on the primary in the format: {"slot_name": int}
-    :param failsafe: failsafe topology. Node is allowed to become the leader only if its name is found in this list.
-    :param workers: workers of the Citus cluster, optional. Format: {int(group): Cluster()}
+
+    :ivar initialize: shows whether this cluster has initialization key stored in DC or not.
+    :ivar config: global dynamic configuration, reference to `ClusterConfig` object.
+    :ivar leader: :class:`Leader` object which represents current leader of the cluster.
+    :ivar last_lsn: :class:int object containing position of last known leader LSN.
+                     This value is stored in the `/status` key or `/optime/leader` (legacy) key.
+    :ivar members: list of:class:` Member` objects, all PostgreSQL cluster members including leader
+    :ivar failover: reference to :class:`Failover` object.
+    :ivar sync: reference to :class:`SyncState` object, last observed synchronous replication state.
+    :ivar history: reference to `TimelineHistory` object.
+    :ivar slots: state of permanent logical replication slots on the primary in the format: {"slot_name": int}.
+    :ivar failsafe: failsafe topology. Node is allowed to become the leader only if its name is found in this list.
+    :ivar workers: dictionary of workers of the Citus cluster, optional. Each key is an :class:`int` representing
+                   the group, and the corresponding value is a :class:`Cluster` instance.
     """
-    initialize: Optional[str]
-    config: Optional[ClusterConfig]
-    leader: Optional[Leader]
-    last_lsn: int
-    members: List[Member]
-    failover: Optional[Failover]
-    sync: SyncState
-    history: Optional[TimelineHistory]
-    slots: Optional[Dict[str, int]]
-    failsafe: Optional[Dict[str, str]]
-    workers: Dict[int, 'Cluster'] = {}
+
+    def __new__(cls, *args: Any, **kwargs: Any):
+        """Make workers argument optional and set it to an empty dict object."""
+        if len(args) < len(cls._fields) and 'workers' not in kwargs:
+            kwargs['workers'] = {}
+        return super(Cluster, cls).__new__(cls, *args, **kwargs)
 
     @staticmethod
     def empty() -> 'Cluster':
+        """Produce an empty :class:`Cluster` instance."""
         return Cluster(None, None, None, 0, [], None, SyncState.empty(), None, None, None)
 
     def is_empty(self):
diff --git a/patroni/file_perm.py b/patroni/file_perm.py
new file mode 100644
index 00000000..ed9c4e67
--- /dev/null
+++ b/patroni/file_perm.py
@@ -0,0 +1,95 @@
+"""Helper object that helps with figuring out file and directory permissions based on permissions of PGDATA.
+
+:var logger: logger of this module.
+:var pg_perm: instance of the :class:`__FilePermissions` object.
+"""
+import logging
+import os
+import stat
+
+logger = logging.getLogger(__name__)
+
+
+class __FilePermissions:
+    """Helper class for managing permissions of directories and files under PGDATA.
+
+    Execute :meth:`set_permissions_from_data_directory` to figure out which permissions should be used for files and
+    directories under PGDATA based on permissions of PGDATA root directory.
+    """
+
+    # Mode mask for data directory permissions that only allows the owner to
+    # read/write directories and files -- mask 077.
+    __PG_MODE_MASK_OWNER = stat.S_IRWXG | stat.S_IRWXO
+
+    # Mode mask for data directory permissions that also allows group read/execute -- mask 027.
+    __PG_MODE_MASK_GROUP = stat.S_IWGRP | stat.S_IRWXO
+
+    # Default mode for creating directories -- mode 700.
+    __PG_DIR_MODE_OWNER = stat.S_IRWXU
+
+    # Mode for creating directories that allows group read/execute -- mode 750.
+    __PG_DIR_MODE_GROUP = stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP
+
+    # Default mode for creating files -- mode 600.
+    __PG_FILE_MODE_OWNER = stat.S_IRUSR | stat.S_IWUSR
+
+    # Mode for creating files that allows group read -- mode 640.
+    __PG_FILE_MODE_GROUP = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP
+
+    def __init__(self) -> None:
+        """Create a :class:`__FilePermissions` object and set default permissions."""
+        self.__set_owner_permissions()
+        self.__set_umask()
+
+    def __set_umask(self) -> None:
+        """Set umask value based on calculations.
+
+        .. note::
+            Should only be called once either :meth:`__set_owner_permissions`
+            or :meth:`__set_group_permissions` has been executed.
+        """
+        try:
+            os.umask(self.__pg_mode_mask)
+        except Exception as e:
+            logger.error('Can not set umask to %03o: %r', self.__pg_mode_mask, e)
+
+    def __set_owner_permissions(self) -> None:
+        """Make directories/files accessible only by the owner."""
+        self.__pg_dir_create_mode = self.__PG_DIR_MODE_OWNER
+        self.__pg_file_create_mode = self.__PG_FILE_MODE_OWNER
+        self.__pg_mode_mask = self.__PG_MODE_MASK_OWNER
+
+    def __set_group_permissions(self) -> None:
+        """Make directories/files accessible by the owner and readable by group."""
+        self.__pg_dir_create_mode = self.__PG_DIR_MODE_GROUP
+        self.__pg_file_create_mode = self.__PG_FILE_MODE_GROUP
+        self.__pg_mode_mask = self.__PG_MODE_MASK_GROUP
+
+    def set_permissions_from_data_directory(self, data_dir: str) -> None:
+        """Set new permissions based on provided *data_dir*.
+
+        :param data_dir: reference to PGDATA to calculate permissions from.
+        """
+        try:
+            st = os.stat(data_dir)
+            if (st.st_mode & self.__PG_DIR_MODE_GROUP) == self.__PG_DIR_MODE_GROUP:
+                self.__set_group_permissions()
+            else:
+                self.__set_owner_permissions()
+        except Exception as e:
+            logger.error('Can not check permissions on %s: %r', data_dir, e)
+        else:
+            self.__set_umask()
+
+    @property
+    def dir_create_mode(self) -> int:
+        """Directory permissions."""
+        return self.__pg_dir_create_mode
+
+    @property
+    def file_create_mode(self) -> int:
+        """File permissions."""
+        return self.__pg_file_create_mode
+
+
+pg_perm = __FilePermissions()
diff --git a/patroni/ha.py b/patroni/ha.py
index 2445aa02..994db00b 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -527,10 +527,17 @@ class Ha(object):
         return msg
 
     def _get_node_to_follow(self, cluster: Cluster) -> Union[Leader, Member, None]:
-        # determine the node to follow. If replicatefrom tag is set,
-        # try to follow the node mentioned there, otherwise, follow the leader.
-        if self.is_standby_cluster() and (self.cluster.is_unlocked() or self.has_lock(False)):
+        """Determine the node to follow.
+
+        :param cluster: the currently known cluster state from DCS.
+
+        :returns: the node which we should be replicating from.
+        """
+        # The standby leader or when there is no standby leader we want to follow
+        # the remote member, except when there is no standby leader in pause.
+        if self.is_standby_cluster() and (self.has_lock(False) or self.cluster.is_unlocked() and not self.is_paused()):
             node_to_follow = self.get_remote_member()
+        # If replicatefrom tag is set, try to follow the node mentioned there, otherwise, follow the leader.
         elif self.patroni.replicatefrom and self.patroni.replicatefrom != self.state_handler.name:
             node_to_follow = cluster.get_member(self.patroni.replicatefrom)
         else:
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index 04fab0a2..e9844710 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -12,7 +12,7 @@ from datetime import datetime
 from dateutil import tz
 from psutil import TimeoutExpired
 from threading import current_thread, Lock
-from typing import Any, Callable, Dict, Generator, List, Optional, Union, Tuple, TYPE_CHECKING
+from typing import Any, Callable, Dict, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING
 
 from .bootstrap import Bootstrap
 from .callback_executor import CallbackAction, CallbackExecutor
@@ -999,7 +999,7 @@ class Postgresql(object):
 
     @contextmanager
     def get_replication_connection_cursor(self, host: Optional[str] = None, port: int = 5432,
-                                          **kwargs: Any) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]:
+                                          **kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]:
         conn_kwargs = self.config.replication.copy()
         conn_kwargs.update(host=host, port=int(port) if port else None, user=conn_kwargs.pop('username'),
                            connect_timeout=3, replication=1, options='-c statement_timeout=2000')
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index 3a61a31d..6d1fe384 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -6,14 +6,16 @@ import socket
 import stat
 import time
 
+from contextlib import contextmanager
 from urllib.parse import urlparse, parse_qsl, unquote
 from types import TracebackType
-from typing import Any, Collection, Dict, List, Optional, Union, Tuple, Type, TYPE_CHECKING
+from typing import Any, Collection, Dict, Iterator, List, Optional, Union, Tuple, Type, TYPE_CHECKING
 
 from .validator import recovery_parameters, transform_postgresql_parameter_value, transform_recovery_parameter_value
 from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
 from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name
 from ..exceptions import PatroniFatalException
+from ..file_perm import pg_perm
 from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, validate_directory, is_subpath
 from ..validator import IntValidator, EnumValidator
 
@@ -367,6 +369,30 @@ class ConfigHandler(object):
             configuration.append('pg_ident.conf')
         return configuration
 
+    def set_file_permissions(self, filename: str) -> None:
+        """Set permissions of file *filename* according to the expected permissions if it resides under PGDATA.
+
+        .. note::
+            Do nothing if the file is not under PGDATA.
+
+        :param filename: path to a file which permissions might need to be adjusted.
+        """
+        if is_subpath(self._postgresql.data_dir, filename):
+            pg_perm.set_permissions_from_data_directory(self._postgresql.data_dir)
+            os.chmod(filename, pg_perm.file_create_mode)
+
+    @contextmanager
+    def config_writer(self, filename: str) -> Iterator[ConfigWriter]:
+        """Create :class:`ConfigWriter` object and set permissions on a *filename*.
+
+        :param filename: path to a config file.
+
+        :yields: :class:`ConfigWriter` object.
+        """
+        with ConfigWriter(filename) as writer:
+            yield writer
+        self.set_file_permissions(filename)
+
     def save_configuration_files(self, check_custom_bootstrap: bool = False) -> bool:
         """
             copy postgresql.conf to postgresql.conf.backup to be able to retrieve configuration files
@@ -380,6 +406,7 @@ class ConfigHandler(object):
                     backup_file = os.path.join(self._postgresql.data_dir, f + '.backup')
                     if os.path.isfile(config_file):
                         shutil.copy(config_file, backup_file)
+                        self.set_file_permissions(backup_file)
             except IOError:
                 logger.exception('unable to create backup copies of configuration files')
         return True
@@ -393,9 +420,11 @@ class ConfigHandler(object):
                 if not os.path.isfile(config_file):
                     if os.path.isfile(backup_file):
                         shutil.copy(backup_file, config_file)
+                        self.set_file_permissions(config_file)
                     # Previously we didn't backup pg_ident.conf, if file is missing just create empty
                     elif f == 'pg_ident.conf':
                         open(config_file, 'w').close()
+                        self.set_file_permissions(config_file)
         except IOError:
             logger.exception('unable to restore configuration files from backup')
 
@@ -409,7 +438,7 @@ class ConfigHandler(object):
         if self._postgresql.enforce_hot_standby_feedback:
             configuration['hot_standby_feedback'] = 'on'
 
-        with ConfigWriter(self._postgresql_conf) as f:
+        with self.config_writer(self._postgresql_conf) as f:
             include = self._config.get('custom_conf') or self._postgresql_base_conf_name
             f.writeline("include '{0}'\n".format(ConfigWriter.escape(include)))
             for name, value in sorted((configuration).items()):
@@ -439,6 +468,7 @@ class ConfigHandler(object):
         if not self.hba_file and not self._config.get('pg_hba'):
             with open(self._pg_hba_conf, 'a') as f:
                 f.write('\n{}\n'.format('\n'.join(config)))
+            self.set_file_permissions(self._pg_hba_conf)
         return True
 
     def replace_pg_hba(self) -> Optional[bool]:
@@ -458,14 +488,14 @@ class ConfigHandler(object):
                                   self.local_replication_address['host'], self.local_replication_address['port'],
                                   0, socket.SOCK_STREAM, socket.IPPROTO_TCP)})
 
-            with ConfigWriter(self._pg_hba_conf) as f:
+            with self.config_writer(self._pg_hba_conf) as f:
                 for address, t in addresses.items():
                     f.writeline((
                         '{0}\treplication\t{1}\t{3}\ttrust\n'
                         '{0}\tall\t{2}\t{3}\ttrust'
                     ).format(t, self.replication['username'], self._superuser.get('username') or 'all', address))
         elif not self.hba_file and self._config.get('pg_hba'):
-            with ConfigWriter(self._pg_hba_conf) as f:
+            with self.config_writer(self._pg_hba_conf) as f:
                 f.writelines(self._config['pg_hba'])
             return True
 
@@ -478,7 +508,7 @@ class ConfigHandler(object):
         """
 
         if not self.ident_file and self._config.get('pg_ident'):
-            with ConfigWriter(self._pg_ident_conf) as f:
+            with self.config_writer(self._pg_ident_conf) as f:
                 f.writelines(self._config['pg_ident'])
             return True
 
@@ -800,9 +830,11 @@ class ConfigHandler(object):
         if self._postgresql.major_version >= 120000:
             if parse_bool(recovery_params.pop('standby_mode', None)):
                 open(self._standby_signal, 'w').close()
+                self.set_file_permissions(self._standby_signal)
             else:
                 self._remove_file_if_exists(self._standby_signal)
                 open(self._recovery_signal, 'w').close()
+                self.set_file_permissions(self._recovery_signal)
 
             def restart_required(name: str) -> bool:
                 if self._postgresql.major_version >= 140000:
@@ -813,8 +845,7 @@ class ConfigHandler(object):
             self._current_recovery_params = CaseInsensitiveDict({n: [v, restart_required(n), self._postgresql_conf]
                                                                  for n, v in recovery_params.items()})
         else:
-            with ConfigWriter(self._recovery_conf) as f:
-                os.chmod(self._recovery_conf, stat.S_IWRITE | stat.S_IREAD)
+            with self.config_writer(self._recovery_conf) as f:
                 self._write_recovery_params(f, recovery_params)
 
     def remove_recovery_conf(self) -> None:
@@ -843,6 +874,7 @@ class ConfigHandler(object):
         if overwrite:
             try:
                 with open(self._auto_conf, 'w') as f:
+                    self.set_file_permissions(self._auto_conf)
                     for raw_line in lines:
                         f.write(raw_line)
             except Exception:
diff --git a/patroni/postgresql/connection.py b/patroni/postgresql/connection.py
index 6a556983..277a4889 100644
--- a/patroni/postgresql/connection.py
+++ b/patroni/postgresql/connection.py
@@ -2,7 +2,7 @@ import logging
 
 from contextlib import contextmanager
 from threading import Lock
-from typing import Any, Dict, Generator, Union, TYPE_CHECKING
+from typing import Any, Dict, Iterator, Union, TYPE_CHECKING
 if TYPE_CHECKING:  # pragma: no cover
     from psycopg import Connection as Connection3, Cursor
     from psycopg2 import connection, cursor
@@ -44,7 +44,7 @@ class Connection(object):
 
 
 @contextmanager
-def get_connection_cursor(**kwargs: Any) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]:
+def get_connection_cursor(**kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]:
     conn = psycopg.connect(**kwargs)
     with conn.cursor() as cur:
         yield cur
diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py
index fde81a7a..82b83002 100644
--- a/patroni/postgresql/slots.py
+++ b/patroni/postgresql/slots.py
@@ -5,11 +5,12 @@ import shutil
 from collections import defaultdict
 from contextlib import contextmanager
 from threading import Condition, Thread
-from typing import Any, Dict, Generator, List, Optional, Union, Tuple, TYPE_CHECKING
+from typing import Any, Dict, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING
 
 from .connection import get_connection_cursor
 from .misc import format_lsn, fsync_dir
 from ..dcs import Cluster, Leader
+from ..file_perm import pg_perm
 from ..psycopg import OperationalError
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -253,7 +254,7 @@ class SlotsHandler(object):
                 self._schedule_load_slots = True
 
     @contextmanager
-    def get_local_connection_cursor(self, **kwargs: Any) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]:
+    def get_local_connection_cursor(self, **kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]:
         conn_kwargs = self._postgresql.config.local_connect_kwargs
         conn_kwargs.update(kwargs)
         with get_connection_cursor(**conn_kwargs) as cur:
@@ -342,7 +343,7 @@ class SlotsHandler(object):
         return ret
 
     @contextmanager
-    def _get_leader_connection_cursor(self, leader: Leader) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]:
+    def _get_leader_connection_cursor(self, leader: Leader) -> Iterator[Union['cursor', 'Cursor[Any]']]:
         conn_kwargs = leader.conn_kwargs(self._postgresql.config.rewind_credentials)
         conn_kwargs['dbname'] = self._postgresql.database
         with get_connection_cursor(connect_timeout=3, options="-c statement_timeout=2000", **conn_kwargs) as cur:
@@ -471,23 +472,28 @@ class SlotsHandler(object):
                 logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e)
 
         if copy_slots and self._postgresql.stop():
+            pg_perm.set_permissions_from_data_directory(self._postgresql.data_dir)
             for name, value in copy_slots.items():
-                slot_dir = os.path.join(self._postgresql.slots_handler.pg_replslot_dir, name)
+                slot_dir = os.path.join(self.pg_replslot_dir, name)
                 slot_tmp_dir = slot_dir + '.tmp'
                 if os.path.exists(slot_tmp_dir):
                     shutil.rmtree(slot_tmp_dir)
                 os.makedirs(slot_tmp_dir)
+                os.chmod(slot_tmp_dir, pg_perm.dir_create_mode)
                 fsync_dir(slot_tmp_dir)
-                with open(os.path.join(slot_tmp_dir, 'state'), 'wb') as f:
+                slot_filename = os.path.join(slot_tmp_dir, 'state')
+                with open(slot_filename, 'wb') as f:
+                    os.chmod(slot_filename, pg_perm.file_create_mode)
                     f.write(value['data'])
                     f.flush()
                     os.fsync(f.fileno())
                 if os.path.exists(slot_dir):
                     shutil.rmtree(slot_dir)
                 os.rename(slot_tmp_dir, slot_dir)
+                os.chmod(slot_dir, pg_perm.dir_create_mode)
                 fsync_dir(slot_dir)
                 self._logical_slots_processing_queue[name] = None
-            fsync_dir(self._postgresql.slots_handler.pg_replslot_dir)
+            fsync_dir(self.pg_replslot_dir)
             self._postgresql.start()
 
     def schedule(self, value: Optional[bool] = None) -> None:


Test patch:
diff --git a/tests/test_config.py b/tests/test_config.py
index 3f7e4049..cf798d00 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -85,6 +85,7 @@ class TestConfig(unittest.TestCase):
     @patch('os.path.exists', Mock(return_value=True))
     @patch('os.remove', Mock(side_effect=IOError))
     @patch('os.close', Mock(side_effect=IOError))
+    @patch('os.chmod', Mock())
     @patch('shutil.move', Mock(return_value=None))
     @patch('json.dump', Mock())
     def test_save_cache(self):
diff --git a/tests/test_file_perm.py b/tests/test_file_perm.py
new file mode 100644
index 00000000..e85b3c65
--- /dev/null
+++ b/tests/test_file_perm.py
@@ -0,0 +1,33 @@
+import unittest
+import stat
+
+from mock import Mock, patch
+
+from patroni.file_perm import pg_perm
+
+
+class TestFilePermissions(unittest.TestCase):
+
+    @patch('os.stat')
+    @patch('os.umask')
+    @patch('patroni.file_perm.logger.error')
+    def test_set_umask(self, mock_logger, mock_umask, mock_stat):
+        mock_umask.side_effect = Exception
+        mock_stat.return_value.st_mode = stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP
+        pg_perm.set_permissions_from_data_directory('test')
+
+        # umask is called with PG_MODE_MASK_GROUP
+        self.assertEqual(mock_umask.call_args[0][0], stat.S_IWGRP | stat.S_IRWXO)
+        self.assertEqual(mock_logger.call_args[0][0], 'Can not set umask to %03o: %r')
+
+        mock_umask.reset_mock()
+        mock_stat.return_value.st_mode = stat.S_IRWXU
+        pg_perm.set_permissions_from_data_directory('test')
+        # umask is called with PG_MODE_MASK_OWNER (permissions changed from group to owner)
+        self.assertEqual(mock_umask.call_args[0][0], stat.S_IRWXG | stat.S_IRWXO)
+
+    @patch('os.stat', Mock(side_effect=FileNotFoundError))
+    @patch('patroni.file_perm.logger.error')
+    def test_set_permissions_from_data_directory(self, mock_logger):
+        pg_perm.set_permissions_from_data_directory('test')
+        self.assertEqual(mock_logger.call_args[0][0], 'Can not check permissions on %s: %r')
diff --git a/tests/test_ha.py b/tests/test_ha.py
index ae6be3e2..d3a75ed3 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -1421,6 +1421,7 @@ class TestHa(PostgresInit):
     @patch('os.open', Mock())
     @patch('os.fsync', Mock())
     @patch('os.close', Mock())
+    @patch('os.chmod', Mock())
     @patch('os.rename', Mock())
     @patch('patroni.postgresql.Postgresql.is_starting', Mock(return_value=False))
     @patch('builtins.open', mock_open())
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 036e225b..b584f2b2 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -523,8 +523,9 @@ class TestPostgresql(BaseTestPostgresql):
     def test_save_configuration_files(self):
         self.p.config.save_configuration_files()
 
-    @patch('os.path.isfile', Mock(side_effect=[False, True]))
-    @patch('shutil.copy', Mock(side_effect=IOError))
+    @patch('os.path.isfile', Mock(side_effect=[False, True, False, True]))
+    @patch('shutil.copy', Mock(side_effect=[None, IOError]))
+    @patch('os.chmod', Mock())
     def test_restore_configuration_files(self):
         self.p.config.restore_configuration_files()