feat: Create consumer cert & key owner by rhsm group

* Card ID: CCT-652
* When consumer certificate or key is created, then try to use
  rhsm group as a owner. When this group does not exist, then
  0 guid is used
* Modified also function checking and correcting permissions
  to consumer cert & key
* Not added any unit test for this case, because it would require
  to run unit tests as a root user

src/subscription_manager/identity.py

@@ -14,6 +14,7 @@
 
 import logging
 import os
+import grp
 import errno
 import threading
 from typing import Optional, TYPE_CHECKING
@@ -108,14 +109,33 @@ def getConsumerOwner(self) -> Optional[str]:
     # TODO: we're using a Certificate which has it's own write/delete, no idea
     # why this landed in a parallel disjoint class wrapping the actual cert.
     def write(self) -> None:
+        """
+        Write consumer key and certificate to disk.
+        """
         from subscription_manager import managerlib
 
+        rhsm_group = None
+        try:
+            rhsm_group = grp.getgrnam(managerlib.RHSM_GROUP_NAME)
+        except KeyError:
+            log.error(f"Unable to get information about {managerlib.RHSM_GROUP_NAME}")
+
         self.__mkdir()
+
         with open(self.keypath(), "w") as key_file:
             key_file.write(self.key)
+
+        # Set proper access permission to the key
+        if os.getuid() == 0 and rhsm_group is not None:
+            os.chown(self.keypath(), 0, rhsm_group.gr_gid)
         os.chmod(self.keypath(), managerlib.ID_CERT_PERMS)
+
         with open(self.certpath(), "w") as cert_file:
             cert_file.write(self.cert)
+
+        # Set proper permission to consumer certificate
+        if os.getuid() == 0 and rhsm_group is not None:
+            os.chown(self.certpath(), 0, rhsm_group.gr_gid)
         os.chmod(self.certpath(), managerlib.ID_CERT_PERMS)
 
     def delete(self) -> None:

src/subscription_manager/managerlib.py

@@ -17,6 +17,7 @@
 import glob
 import logging
 import os
+import grp
 import re
 import shutil
 import stat
@@ -75,6 +76,7 @@
 
 # Expected permissions for identity certificates:
 ID_CERT_PERMS: int = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP
+RHSM_GROUP_NAME: str = "rhsm"
 
 
 def system_log(message: str, priority: int = syslog.LOG_NOTICE) -> None:
@@ -934,26 +936,42 @@ def format_iso8601_date(dateobj: Optional[datetime.datetime]) -> str:
     return ""
 
 
+def get_rhsm_group() -> Optional[grp.struct_group]:
+    """
+    Try to get GUID about rhsm group
+    """
+    rhsm_group = None
+    try:
+        rhsm_group = grp.getgrnam(RHSM_GROUP_NAME)
+    except KeyError:
+        log.error(f"Unable to get information about {RHSM_GROUP_NAME}")
+    return rhsm_group
+
+
 # FIXME: move me to identity.py
 def check_identity_cert_perms() -> None:
     """
     Ensure the identity certs on this system have the correct permissions, and
     fix them if not.
     """
     certs: List[str] = [identity.ConsumerIdentity.keypath(), identity.ConsumerIdentity.certpath()]
+    rhsm_group = get_rhsm_group()
+    cert_guid = 0
+    if rhsm_group is not None:
+        cert_guid = rhsm_group.gr_gid
     for cert in certs:
         if not os.path.exists(cert):
             # Only relevant if these files exist.
             continue
         statinfo: os.stat_result = os.stat(cert)
-        if statinfo[stat.ST_UID] != 0 or statinfo[stat.ST_GID] != 0:
-            os.chown(cert, 0, 0)
-            log.warn("Corrected incorrect ownership of %s." % cert)
+        if statinfo[stat.ST_UID] != 0 or statinfo[stat.ST_GID] != cert_guid:
+            os.chown(cert, 0, cert_guid)
+            log.warning("Corrected incorrect ownership of %s." % cert)
 
         mode: int = stat.S_IMODE(statinfo[stat.ST_MODE])
         if mode != ID_CERT_PERMS:
             os.chmod(cert, ID_CERT_PERMS)
-            log.warn("Corrected incorrect permissions on %s." % cert)
+            log.warning("Corrected incorrect permissions on %s." % cert)
 
 
 def clean_all_data(backup: bool = True) -> None: