12583462520

Committed 02 Jan 2025 01:51PM CUT coverage: 78.273%. Remained the same

Build # 12583462520

Build Type

Pull #6206

github

Committed by

web-flow

Commit Message

Merge 75f0b41f9 into 5f6b50041

Pull Request Pull Request #6206: xenopsd: Avoid calling to_string every time

Run Details

3462 of 4423 relevant lines covered (78.27%)

0.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.87

/python3/plugins/extauth-hook-AD.py

#!/usr/bin/env python3
#
# extauth-hook-AD.py
#
# This plugin manages the following configuration files for external authentication
# - /etc/nsswitch.conf
# - /etc/pam.d/sshd
# - /etc/pam.d/hcp_users
# - /etc/ssh/ssh_config
#
# This module can be called directly as a plugin.  It handles
# Active Directory being enabled or disabled as the hosts external_auth_type,
# or subjects being added or removed while AD is the external_auth_type,
# or xapi starting or stopping while AD is the external_auth_type.
#
# Alternatively, the extauth-hook module can be called, which will
# dispatch to the correct extauth-hook-<type>.py module automatically.
import abc
import subprocess
import os
import shutil
import tempfile
import logging
import logging.handlers
from collections import OrderedDict

import XenAPIPlugin


# pylint: disable=too-few-public-methods


HCP_USERS = "/etc/security/hcp_ad_users.conf"
HCP_GROUPS = "/etc/security/hcp_ad_groups.conf"


def setup_logger():
    """Helper function setup logger"""
    addr = "/dev/log"

    logging.basicConfig(
        format='%(asctime)s %(levelname)s %(name)s %(funcName)s %(message)s', level=logging.DEBUG)
    log = logging.getLogger()

    if not os.path.exists(addr):
        log.warning("%s not available, logs are not redirected", addr)
        return

    # Send to syslog local5, which will be redirected to xapi log /var/log/xensource.log
    handler = logging.handlers.SysLogHandler(
        facility='local5', address=addr)
    # Send to authpriv, which will be redirected to /var/log/secure
    auth_log = logging.handlers.SysLogHandler(
        facility="authpriv", address=addr)
    log.addHandler(handler)
    log.addHandler(auth_log)


setup_logger()
logger = logging.getLogger(__name__)


def run_cmd(command: "list[str]"):
    """Helper function to run a command and log the output"""
    try:
        output = subprocess.check_output(command, universal_newlines=True)
        logger.debug("%s -> %s", command, output.strip())

    except OSError:
        logger.exception("Failed to run command %s", command)


class ADConfig(abc.ABC):
    """Base class for AD configuration"""

    def __init__(self, path, session, args, ad_enabled=True, load_existing=True, file_mode=0o644):
        self._file_path = path
        self._session = session
        self._args = args
        self._lines = []
        self._ad_enabled = ad_enabled
        self._file_mode = file_mode
        if load_existing and os.path.exists(self._file_path):
            with open(self._file_path, "r", encoding="utf-8") as file:
                lines = file.readlines()
                self._lines = [l.strip() for l in lines]


    @abc.abstractmethod
    def _apply_to_cache(self): ...

    def apply(self):
        """Apply configuration"""
        self._apply_to_cache()
        self._install()

    def _install(self):
        """Install configuration"""
        with tempfile.NamedTemporaryFile(prefix="extauth-", delete=False) as file:
            file.write("\n".join(self._lines).encode("utf-8"))
            file.flush()
            shutil.move(file.name, self._file_path)
            os.chmod(self._file_path, self._file_mode)


class StaticSSHPam(ADConfig):
    """
    Class to manage ssh pam configuration
    """
    ad_pam_format = """#%PAM-1.0
auth        required      pam_env.so
auth        sufficient    pam_unix.so try_first_pass nullok
auth        sufficient    {ad_module} try_first_pass try_authtok
auth        required      pam_deny.so

session     optional      pam_keyinit.so force revoke
session     required      pam_limits.so
session     [success=1 default=ignore] pam_succeed_if.so service in crond quiet use_uid
session     required      pam_unix.so
session     required      pam_loginuid.so
session     sufficient    {ad_module}

account     required      pam_nologin.so
account     required      {ad_module} unknown_ok
account     sufficient    pam_listfile.so item=user file={user_list} sense=allow onerr=fail
account     sufficient    pam_listfile.so item=group file={group_list} sense=allow onerr=fail
account     required      pam_unix.so"""

    no_ad_pam = """#%PAM-1.0
auth       include      system-auth
account    required     pam_nologin.so
account    include      system-auth
password   include      system-auth
session    optional     pam_keyinit.so force revoke
session    include      system-auth
session    required     pam_loginuid.so"""

    def __init__(self, session, args, ad_enabled=True):
        super(StaticSSHPam, self).__init__("/etc/pam.d/sshd", session, args, ad_enabled,
                                           load_existing=False)

    def _apply_to_cache(self):
        if self._ad_enabled:
            content = self.ad_pam_format.format(ad_module="pam_winbind.so",
                                                user_list=HCP_USERS, group_list=HCP_GROUPS)
        else:
            content = self.no_ad_pam
        self._lines = content.split("\n")


class DynamicPam(ADConfig):
    """Base class to manage AD users and groups configure which permit pool admin ssh"""

    def __init__(self, path, session, args, ad_enabled=True):
        super(DynamicPam, self).__init__(path, session, args, ad_enabled,
                                         load_existing=False)
        self.admin_role = self._session.xenapi.role.get_by_name_label(
            'pool-admin')[0]

    def _apply_to_cache(self):
        # AD is not enabled, just return as the configure file will be removed during install
        if not self._ad_enabled:
            return

        try:
            # Rewrite the PAM SSH config using the latest info from Active Directory
            # and the list of subjects from xapi
            subjects = self._session.xenapi.subject.get_all()
            # Add each subject which contains the admin role
            for opaque_ref in subjects:
                subject_rec = self._session.xenapi.subject.get_record(
                    opaque_ref)
                if self._is_pool_admin(subject_rec) and self._is_responsible_for(subject_rec):
                    self._add_subject(subject_rec)
            self._lines.append("")  # ending new line
        except Exception as exp:  # pylint: disable=broad-except
            logger.info("Failed to add subjects %s", str(exp))

    def _is_pool_admin(self, subject_rec):
        try:
            return self.admin_role in subject_rec['roles']
        except KeyError:
            logger.warning("subject %s does not have role", subject_rec)
            return False


    def _is_responsible_for(self, subject_rec):
        try:
            return self._match_subject(subject_rec)
        except KeyError:
            logger.exception("Failed to match subject %s", subject_rec)
            return False

    @abc.abstractmethod
    def _match_subject(self, subject_rec): ...

    @abc.abstractmethod
    def _add_subject(self, subject_rec): ...

    def _install(self):
        if self._ad_enabled:
            super(DynamicPam, self)._install()
        elif os.path.exists(self._file_path):
            os.remove(self._file_path)


class UsersList(DynamicPam):
    """Class manage users which permit pool admin ssh"""

    def __init__(self, session, arg, ad_enabled=True):
        super(UsersList, self).__init__(HCP_USERS, session, arg, ad_enabled)

    def _match_subject(self, subject_rec):
        return subject_rec["other_config"]["subject-is-group"] != "true"

    def _add_upn(self, subject_rec):
        sep = "@"
        upn = ""
        try:
            upn = subject_rec["other_config"]["subject-upn"]
            user, domain = upn.split(sep)
            self._lines.append("{}{}{}".format(user, sep, domain))
        except KeyError:
            logger.info("subject does not have upn %s", subject_rec)
        except ValueError:
            logger.info("UPN format is not right %s", upn)

    def _add_subject(self, subject_rec):
        try:
            sid = subject_rec['subject_identifier']
            formatted_name = subject_rec["other_config"]["subject-name"]
            logger.debug("Permit user %s, Current sid is %s",
                         formatted_name, sid)
            self._lines.append(formatted_name)
            # If the ssh key is permitted in the authorized_keys file,
            # The original name is compared, add UPN and original name
            self._add_upn(subject_rec)
        # pylint: disable=broad-except
        except Exception as exp:
            logger.warning("Failed to add user %s: %s", subject_rec, str(exp))


class GroupsList(DynamicPam):
    """Class manage groups which permit pool admin ssh"""

    def __init__(self, session, arg, ad_enabled=True):
        super(GroupsList, self).__init__(HCP_GROUPS, session, arg, ad_enabled)

    def _match_subject(self, subject_rec):
        return subject_rec["other_config"]["subject-is-group"] == "true"

    def _add_subject(self, subject_rec):
        try:
            sid = subject_rec['subject_identifier']
            name = subject_rec["other_config"]["subject-name"]
            logger.debug("Permit group %s, Current sid is %s", name, sid)
            self._lines.append(name)
       # pylint: disable=broad-except
        except Exception as exp:
            logger.warning("Failed to add group %s:%s", subject_rec, str(exp))


class KeyValueConfig(ADConfig):
    """
     Only support configure files with key value in each line, separated by sep
     Otherwise, it will be just copied and un-configurable
     If multiple lines with the same key exists, only the first line will be configured
    """
    # Presume normal config does not have such keys
    _special_line_prefix = "__key_value_config_sp_line_prefix_"
    _empty_value = ""

    def __init__(self, path, session, args, ad_enabled=True, load_existing=True,
                 file_mode=0o644, sep=": ", comment="#"):
        super(KeyValueConfig, self).__init__(path, session,
                                             args, ad_enabled, load_existing, file_mode)
        self._sep = None if sep.isspace() else sep  # Ignore number/type of spaces
        self._comment = comment
        self._values = OrderedDict()
        self._load_values()

    def _is_comment_line(self, line):
        return line.startswith(self._comment)

    def _is_special_line(self, line):
        """Whether the line is a special line"""
        return line.startswith(self._special_line_prefix)

    def _load_values(self):
        for idx, line in enumerate(self._lines):
            # Generate a unique key to store multiple special lines
            sp_key = "{}{}".format(self._special_line_prefix, str(idx))
            if line == "":  # Empty line
                self._values[sp_key] = self._empty_value
            elif self._is_comment_line(line):
                self._values[sp_key] = line
            else:  # Parse the key, value pair
                item = line.split(self._sep)
                if len(item) != 2:
                    # Taken as raw line
                    self._values[sp_key] = line
                else:
                    key, value = item[0].strip(), item[1].strip()
                    if key not in self._values:
                        self._values[key] = value
                    else:
                        # Key already exists, Not supported as configurable
                        self._values[sp_key] = line

    def _update_key_value(self, key, value):
        self._values[key] = value

    def _apply_value(self, key, value):
        if self._is_special_line(key):
            line = value
        else:  # normal line, construct the key value pair
            sep = self._sep or " "
            line = "{}{}{}".format(key, sep, value)
        self._lines.append(line)

    def _apply_to_cache(self):
        self._lines = []
        for key, value in self._values.items():
            self._apply_value(key, value)


class NssConfig(KeyValueConfig):
    """Class to manage NSS configuration"""

    def __init__(self, session, args, ad_enabled=True):
        super(NssConfig, self).__init__(
            "/etc/nsswitch.conf", session, args, ad_enabled)
        modules = "files sss"
        if ad_enabled:
            modules = "files hcp winbind"
        self._update_key_value("passwd", modules)
        self._update_key_value("group", modules)
        self._update_key_value("shadow", modules)


class SshdConfig(KeyValueConfig):
    """Class to manage sshd configuration"""

    def __init__(self, session, args, ad_enabled=True):
        super(SshdConfig, self).__init__("/etc/ssh/sshd_config", session, args, ad_enabled,
                                         sep=" ")
        value = "yes" if ad_enabled else "no"
        self._update_key_value("ChallengeResponseAuthentication", value)
        self._update_key_value("GSSAPIAuthentication", value)
        self._update_key_value("GSSAPICleanupCredentials", value)

    def apply(self):
        super(SshdConfig, self).apply()
        run_cmd(["/usr/bin/systemctl", "reload-or-restart", "sshd"])


class PamWinbindConfig(KeyValueConfig):
    """Class to manage winbind pam configuration"""

    def __init__(self, session, args, ad_enabled=True):
        super(PamWinbindConfig, self).__init__("/etc/security/pam_winbind.conf", session, args,
                                               ad_enabled, sep=" = ")
        self._update_key_value("krb5_auth", "yes")


class ConfigManager:
    """Class to manage all the AD configurations"""

    def __init__(self, session, args, ad_enabled=True):
        self._build_config(session, args, ad_enabled)

    def _build_config(self, session, args, ad_enabled):
        self._nss = NssConfig(session, args, ad_enabled)
        self._sshd = SshdConfig(session, args, ad_enabled)
        self._static_pam = StaticSSHPam(session, args, ad_enabled)
        self._users = UsersList(session, args, ad_enabled)
        self._groups = GroupsList(session, args, ad_enabled)
        self._pam_winbind = PamWinbindConfig(session, args, ad_enabled)

    def refresh_all(self):
        """Update all the configurations"""
        self._nss.apply()
        self._sshd.apply()
        self._users.apply()
        self._groups.apply()
        self._static_pam.apply()
        self._pam_winbind.apply()

    def refresh_dynamic_pam(self):
        """Only refresh the dynanic configurations"""
        self._users.apply()
        self._groups.apply()


def refresh_all_configurations(session, args, name, ad_enabled=True):
    """Update all configurations"""
    try:
        logger.info("refresh_all_configurations for %s", name)
        ConfigManager(session, args, ad_enabled).refresh_all()
        return str(True)
    except Exception:  # pylint: disable=broad-except
        msg = "Failed to refresh all configurations"
        logger.exception(msg)
        return msg


def refresh_dynamic_pam(session, args, name):
    """Refresh dynamic pam configurations"""
    try:
        logger.info("refresh_dynamic_pam for %s", name)
        ConfigManager(session, args).refresh_dynamic_pam()
        return str(True)
    except Exception:  # pylint: disable=broad-except
        msg = "Failed to refresh dynamic pam configuration"
        logger.exception(msg)
        return msg


def after_extauth_enable(session, args):
    """Callback for after enable external auth"""
    return refresh_all_configurations(session, args, "after_extauth_enable")


def after_xapi_initialize(session, args):
    """Callback after xapi initialization"""
    return refresh_all_configurations(session, args, "after_xapi_initialize")


def after_subject_add(session, args):
    """Callback after add subject"""
    return refresh_dynamic_pam(session, args, "after_subject_add")


def after_subject_remove(session, args):
    """Callback after remove subject"""
    return refresh_dynamic_pam(session, args, "after_subject_remove")


def after_subject_update(session, args):
    """Callback after subject update"""
    return refresh_dynamic_pam(session, args, "after_subject_update")


def after_roles_update(session, args):
    """Callback after roles update"""
    return refresh_dynamic_pam(session, args, "after_roles_update")


def before_extauth_disable(session, args):
    """Callback before disable external auth"""
    return refresh_all_configurations(session, args, "before_extauth_disable", False)


# The dispatcher
if __name__ == "__main__":
    dispatch_tbl = {
        "after-extauth-enable":   after_extauth_enable,
        "after-xapi-initialize":  after_xapi_initialize,
        "after-subject-add":      after_subject_add,
        "after-subject-update":   after_subject_update,
        "after-subject-remove":   after_subject_remove,
        "after-roles-update":     after_roles_update,
        "before-extauth-disable": before_extauth_disable,
    }
    XenAPIPlugin.dispatch(dispatch_tbl)

1	#!/usr/bin/env python3
2	#
3	# extauth-hook-AD.py
4	#
5	# This plugin manages the following configuration files for external authentication
6	# - /etc/nsswitch.conf
7	# - /etc/pam.d/sshd
8	# - /etc/pam.d/hcp_users
9	# - /etc/ssh/ssh_config
10	#
11	# This module can be called directly as a plugin. It handles
12	# Active Directory being enabled or disabled as the hosts external_auth_type,
13	# or subjects being added or removed while AD is the external_auth_type,
14	# or xapi starting or stopping while AD is the external_auth_type.
15	#
16	# Alternatively, the extauth-hook module can be called, which will
17	# dispatch to the correct extauth-hook-<type>.py module automatically.
18	import abc	1✔
19	import subprocess	1✔
20	import os	1✔
21	import shutil	1✔
22	import tempfile	1✔
23	import logging	1✔
24	import logging.handlers	1✔
25	from collections import OrderedDict	1✔
26
27	import XenAPIPlugin	1✔
28
29
30	# pylint: disable=too-few-public-methods
31
32
33	HCP_USERS = "/etc/security/hcp_ad_users.conf"	1✔
34	HCP_GROUPS = "/etc/security/hcp_ad_groups.conf"	1✔
35
36
37	def setup_logger():	1✔
38	"""Helper function setup logger"""
39	addr = "/dev/log"	1✔
40
41	logging.basicConfig(	1✔
42	format='%(asctime)s %(levelname)s %(name)s %(funcName)s %(message)s', level=logging.DEBUG)
43	log = logging.getLogger()	1✔
44
45	if not os.path.exists(addr):	1✔
46	log.warning("%s not available, logs are not redirected", addr)	×
47	return	×
48
49	# Send to syslog local5, which will be redirected to xapi log /var/log/xensource.log
50	handler = logging.handlers.SysLogHandler(	1✔
51	facility='local5', address=addr)
52	# Send to authpriv, which will be redirected to /var/log/secure
53	auth_log = logging.handlers.SysLogHandler(	1✔
54	facility="authpriv", address=addr)
55	log.addHandler(handler)	1✔
56	log.addHandler(auth_log)	1✔
57
58
59	setup_logger()	1✔
60	logger = logging.getLogger(__name__)	1✔
61
62
63	def run_cmd(command: "list[str]"):	1✔
64	"""Helper function to run a command and log the output"""
65	try:	1✔
66	output = subprocess.check_output(command, universal_newlines=True)	1✔
67	logger.debug("%s -> %s", command, output.strip())	1✔
68
69	except OSError:	1✔
70	logger.exception("Failed to run command %s", command)	1✔
71
72
73	class ADConfig(abc.ABC):	1✔
74	"""Base class for AD configuration"""
75
76	def __init__(self, path, session, args, ad_enabled=True, load_existing=True, file_mode=0o644):	1✔
77	self._file_path = path	1✔
78	self._session = session	1✔
79	self._args = args	1✔
80	self._lines = []	1✔
81	self._ad_enabled = ad_enabled	1✔
82	self._file_mode = file_mode	1✔
83	if load_existing and os.path.exists(self._file_path):	1✔
84	with open(self._file_path, "r", encoding="utf-8") as file:	1✔
85	lines = file.readlines()	1✔
86	self._lines = [l.strip() for l in lines]	1✔
87
88
89	@abc.abstractmethod	1✔
90	def _apply_to_cache(self): ...	1✔
91
92	def apply(self):	1✔
93	"""Apply configuration"""
94	self._apply_to_cache()	1✔
95	self._install()	1✔
96
97	def _install(self):	1✔
98	"""Install configuration"""
99	with tempfile.NamedTemporaryFile(prefix="extauth-", delete=False) as file:	1✔
100	file.write("\n".join(self._lines).encode("utf-8"))	1✔
101	file.flush()	1✔
102	shutil.move(file.name, self._file_path)	1✔
103	os.chmod(self._file_path, self._file_mode)	1✔
104
105
106	class StaticSSHPam(ADConfig):	1✔
107	"""
108	Class to manage ssh pam configuration
109	"""
110	ad_pam_format = """#%PAM-1.0
111	auth required pam_env.so
112	auth sufficient pam_unix.so try_first_pass nullok
113	auth sufficient {ad_module} try_first_pass try_authtok
114	auth required pam_deny.so
115
116	session optional pam_keyinit.so force revoke
117	session required pam_limits.so
118	session [success=1 default=ignore] pam_succeed_if.so service in crond quiet use_uid
119	session required pam_unix.so
120	session required pam_loginuid.so
121	session sufficient {ad_module}
122
123	account required pam_nologin.so
124	account required {ad_module} unknown_ok
125	account sufficient pam_listfile.so item=user file={user_list} sense=allow onerr=fail
126	account sufficient pam_listfile.so item=group file={group_list} sense=allow onerr=fail
127	account required pam_unix.so"""
128
129	no_ad_pam = """#%PAM-1.0
130	auth include system-auth
131	account required pam_nologin.so
132	account include system-auth
133	password include system-auth
134	session optional pam_keyinit.so force revoke
135	session include system-auth
136	session required pam_loginuid.so"""
137
138	def __init__(self, session, args, ad_enabled=True):	1✔
139	super(StaticSSHPam, self).__init__("/etc/pam.d/sshd", session, args, ad_enabled,	1✔
140	load_existing=False)
141
142	def _apply_to_cache(self):	1✔
143	if self._ad_enabled:	1✔
144	content = self.ad_pam_format.format(ad_module="pam_winbind.so",	1✔
145	user_list=HCP_USERS, group_list=HCP_GROUPS)
146	else:
147	content = self.no_ad_pam	1✔
148	self._lines = content.split("\n")	1✔
149
150
151	class DynamicPam(ADConfig):	1✔
152	"""Base class to manage AD users and groups configure which permit pool admin ssh"""
153
154	def __init__(self, path, session, args, ad_enabled=True):	1✔
155	super(DynamicPam, self).__init__(path, session, args, ad_enabled,	1✔
156	load_existing=False)
157	self.admin_role = self._session.xenapi.role.get_by_name_label(	1✔
158	'pool-admin')[0]
159
160	def _apply_to_cache(self):	1✔
161	# AD is not enabled, just return as the configure file will be removed during install
162	if not self._ad_enabled:	1✔
163	return	1✔
164
165	try:	1✔
166	# Rewrite the PAM SSH config using the latest info from Active Directory
167	# and the list of subjects from xapi
168	subjects = self._session.xenapi.subject.get_all()	1✔
169	# Add each subject which contains the admin role
170	for opaque_ref in subjects:	1✔
171	subject_rec = self._session.xenapi.subject.get_record(	1✔
172	opaque_ref)
173	if self._is_pool_admin(subject_rec) and self._is_responsible_for(subject_rec):	1✔
174	self._add_subject(subject_rec)	1✔
175	self._lines.append("") # ending new line	1✔
176	except Exception as exp: # pylint: disable=broad-except	×
177	logger.info("Failed to add subjects %s", str(exp))	×
178
179	def _is_pool_admin(self, subject_rec):	1✔
180	try:	1✔
181	return self.admin_role in subject_rec['roles']	1✔
182	except KeyError:	×
183	logger.warning("subject %s does not have role", subject_rec)	×
184	return False	×
185
186
187	def _is_responsible_for(self, subject_rec):	1✔
188	try:	1✔
189	return self._match_subject(subject_rec)	1✔
190	except KeyError:	×
191	logger.exception("Failed to match subject %s", subject_rec)	×
192	return False	×
193
194	@abc.abstractmethod	1✔
195	def _match_subject(self, subject_rec): ...	1✔
196
197	@abc.abstractmethod	1✔
198	def _add_subject(self, subject_rec): ...	1✔
199
200	def _install(self):	1✔
201	if self._ad_enabled:	1✔
202	super(DynamicPam, self)._install()	1✔
203	elif os.path.exists(self._file_path):	1✔
204	os.remove(self._file_path)	1✔
205
206
207	class UsersList(DynamicPam):	1✔
208	"""Class manage users which permit pool admin ssh"""
209
210	def __init__(self, session, arg, ad_enabled=True):	1✔
211	super(UsersList, self).__init__(HCP_USERS, session, arg, ad_enabled)	1✔
212
213	def _match_subject(self, subject_rec):	1✔
214	return subject_rec["other_config"]["subject-is-group"] != "true"	1✔
215
216	def _add_upn(self, subject_rec):	1✔
217	sep = "@"	1✔
218	upn = ""	1✔
219	try:	1✔
220	upn = subject_rec["other_config"]["subject-upn"]	1✔
221	user, domain = upn.split(sep)	1✔
222	self._lines.append("{}{}{}".format(user, sep, domain))	1✔
223	except KeyError:	×
224	logger.info("subject does not have upn %s", subject_rec)	×
225	except ValueError:	×
226	logger.info("UPN format is not right %s", upn)	×
227
228	def _add_subject(self, subject_rec):	1✔
229	try:	1✔
230	sid = subject_rec['subject_identifier']	1✔
231	formatted_name = subject_rec["other_config"]["subject-name"]	1✔
232	logger.debug("Permit user %s, Current sid is %s",	1✔
233	formatted_name, sid)
234	self._lines.append(formatted_name)	1✔
235	# If the ssh key is permitted in the authorized_keys file,
236	# The original name is compared, add UPN and original name
237	self._add_upn(subject_rec)	1✔
238	# pylint: disable=broad-except
239	except Exception as exp:	×
240	logger.warning("Failed to add user %s: %s", subject_rec, str(exp))	×
241
242
243	class GroupsList(DynamicPam):	1✔
244	"""Class manage groups which permit pool admin ssh"""
245
246	def __init__(self, session, arg, ad_enabled=True):	1✔
247	super(GroupsList, self).__init__(HCP_GROUPS, session, arg, ad_enabled)	1✔
248
249	def _match_subject(self, subject_rec):	1✔
250	return subject_rec["other_config"]["subject-is-group"] == "true"	1✔
251
252	def _add_subject(self, subject_rec):	1✔
253	try:	1✔
254	sid = subject_rec['subject_identifier']	1✔
255	name = subject_rec["other_config"]["subject-name"]	1✔
256	logger.debug("Permit group %s, Current sid is %s", name, sid)	1✔
257	self._lines.append(name)	1✔
258	# pylint: disable=broad-except
259	except Exception as exp:	×
260	logger.warning("Failed to add group %s:%s", subject_rec, str(exp))	×
261
262
263	class KeyValueConfig(ADConfig):	1✔
264	"""
265	Only support configure files with key value in each line, separated by sep
266	Otherwise, it will be just copied and un-configurable
267	If multiple lines with the same key exists, only the first line will be configured
268	"""
269	# Presume normal config does not have such keys
270	_special_line_prefix = "__key_value_config_sp_line_prefix_"	1✔
271	_empty_value = ""	1✔
272
273	def __init__(self, path, session, args, ad_enabled=True, load_existing=True,	1✔
274	file_mode=0o644, sep=": ", comment="#"):
275	super(KeyValueConfig, self).__init__(path, session,	1✔
276	args, ad_enabled, load_existing, file_mode)
277	self._sep = None if sep.isspace() else sep # Ignore number/type of spaces	1✔
278	self._comment = comment	1✔
279	self._values = OrderedDict()	1✔
280	self._load_values()	1✔
281
282	def _is_comment_line(self, line):	1✔
283	return line.startswith(self._comment)	1✔
284
285	def _is_special_line(self, line):	1✔
286	"""Whether the line is a special line"""
287	return line.startswith(self._special_line_prefix)	1✔
288
289	def _load_values(self):	1✔
290	for idx, line in enumerate(self._lines):	1✔
291	# Generate a unique key to store multiple special lines
292	sp_key = "{}{}".format(self._special_line_prefix, str(idx))	1✔
293	if line == "": # Empty line	1✔
294	self._values[sp_key] = self._empty_value	1✔
295	elif self._is_comment_line(line):	1✔
296	self._values[sp_key] = line	1✔
297	else: # Parse the key, value pair
298	item = line.split(self._sep)	1✔
299	if len(item) != 2:	1✔
300	# Taken as raw line
301	self._values[sp_key] = line	×
302	else:
303	key, value = item[0].strip(), item[1].strip()	1✔
304	if key not in self._values:	1✔
305	self._values[key] = value	1✔
306	else:
307	# Key already exists, Not supported as configurable
308	self._values[sp_key] = line	×
309
310	def _update_key_value(self, key, value):	1✔
311	self._values[key] = value	1✔
312
313	def _apply_value(self, key, value):	1✔
314	if self._is_special_line(key):	1✔
315	line = value	1✔
316	else: # normal line, construct the key value pair
317	sep = self._sep or " "	1✔
318	line = "{}{}{}".format(key, sep, value)	1✔
319	self._lines.append(line)	1✔
320
321	def _apply_to_cache(self):	1✔
322	self._lines = []	1✔
323	for key, value in self._values.items():	1✔
324	self._apply_value(key, value)	1✔
325
326
327	class NssConfig(KeyValueConfig):	1✔
328	"""Class to manage NSS configuration"""
329
330	def __init__(self, session, args, ad_enabled=True):	1✔
331	super(NssConfig, self).__init__(	1✔
332	"/etc/nsswitch.conf", session, args, ad_enabled)
333	modules = "files sss"	1✔
334	if ad_enabled:	1✔
335	modules = "files hcp winbind"	1✔
336	self._update_key_value("passwd", modules)
337	self._update_key_value("group", modules)	1✔
338	self._update_key_value("shadow", modules)	1✔
339
340
341	class SshdConfig(KeyValueConfig):	1✔
342	"""Class to manage sshd configuration"""
343
344	def __init__(self, session, args, ad_enabled=True):	1✔
345	super(SshdConfig, self).__init__("/etc/ssh/sshd_config", session, args, ad_enabled,	1✔
346	sep=" ")
347	value = "yes" if ad_enabled else "no"	1✔
348	self._update_key_value("ChallengeResponseAuthentication", value)	1✔
349	self._update_key_value("GSSAPIAuthentication", value)	1✔
350	self._update_key_value("GSSAPICleanupCredentials", value)	1✔
351
352	def apply(self):	1✔
353	super(SshdConfig, self).apply()	1✔
354	run_cmd(["/usr/bin/systemctl", "reload-or-restart", "sshd"])	1✔
355
356
357	class PamWinbindConfig(KeyValueConfig):	1✔
358	"""Class to manage winbind pam configuration"""
359
360	def __init__(self, session, args, ad_enabled=True):	1✔
361	super(PamWinbindConfig, self).__init__("/etc/security/pam_winbind.conf", session, args,	×
362	ad_enabled, sep=" = ")
363	self._update_key_value("krb5_auth", "yes")	×
364
365
366	class ConfigManager:	1✔
367	"""Class to manage all the AD configurations"""
368
369	def __init__(self, session, args, ad_enabled=True):	1✔
370	self._build_config(session, args, ad_enabled)	×
371
372	def _build_config(self, session, args, ad_enabled):	1✔
373	self._nss = NssConfig(session, args, ad_enabled)	×
374	self._sshd = SshdConfig(session, args, ad_enabled)	×
375	self._static_pam = StaticSSHPam(session, args, ad_enabled)	×
376	self._users = UsersList(session, args, ad_enabled)	×
377	self._groups = GroupsList(session, args, ad_enabled)	×
378	self._pam_winbind = PamWinbindConfig(session, args, ad_enabled)	×
379
380	def refresh_all(self):	1✔
381	"""Update all the configurations"""
382	self._nss.apply()	×
383	self._sshd.apply()	×
384	self._users.apply()	×
385	self._groups.apply()	×
386	self._static_pam.apply()	×
387	self._pam_winbind.apply()	×
388
389	def refresh_dynamic_pam(self):	1✔
390	"""Only refresh the dynanic configurations"""
391	self._users.apply()	×
392	self._groups.apply()	×
393
394
395	def refresh_all_configurations(session, args, name, ad_enabled=True):	1✔
396	"""Update all configurations"""
397	try:	×
398	logger.info("refresh_all_configurations for %s", name)	×
399	ConfigManager(session, args, ad_enabled).refresh_all()	×
400	return str(True)	×
401	except Exception: # pylint: disable=broad-except
402	msg = "Failed to refresh all configurations"
403	logger.exception(msg)
404	return msg
405
406
407	def refresh_dynamic_pam(session, args, name):	1✔
408	"""Refresh dynamic pam configurations"""
409	try:	×
410	logger.info("refresh_dynamic_pam for %s", name)	×
411	ConfigManager(session, args).refresh_dynamic_pam()	×
412	return str(True)	×
413	except Exception: # pylint: disable=broad-except
414	msg = "Failed to refresh dynamic pam configuration"
415	logger.exception(msg)
416	return msg
417
418
419	def after_extauth_enable(session, args):	1✔
420	"""Callback for after enable external auth"""
421	return refresh_all_configurations(session, args, "after_extauth_enable")	×
422
423
424	def after_xapi_initialize(session, args):	1✔
425	"""Callback after xapi initialization"""
426	return refresh_all_configurations(session, args, "after_xapi_initialize")	×
427
428
429	def after_subject_add(session, args):	1✔
430	"""Callback after add subject"""
431	return refresh_dynamic_pam(session, args, "after_subject_add")	×
432
433
434	def after_subject_remove(session, args):	1✔
435	"""Callback after remove subject"""
436	return refresh_dynamic_pam(session, args, "after_subject_remove")	×
437
438
439	def after_subject_update(session, args):	1✔
440	"""Callback after subject update"""
441	return refresh_dynamic_pam(session, args, "after_subject_update")	×
442
443
444	def after_roles_update(session, args):	1✔
445	"""Callback after roles update"""
446	return refresh_dynamic_pam(session, args, "after_roles_update")	×
447
448
449	def before_extauth_disable(session, args):	1✔
450	"""Callback before disable external auth"""
451	return refresh_all_configurations(session, args, "before_extauth_disable", False)	×
452
453
454	# The dispatcher
455	if __name__ == "__main__":	1✔
456	dispatch_tbl = {	×
457	"after-extauth-enable": after_extauth_enable,
458	"after-xapi-initialize": after_xapi_initialize,
459	"after-subject-add": after_subject_add,
460	"after-subject-update": after_subject_update,
461	"after-subject-remove": after_subject_remove,
462	"after-roles-update": after_roles_update,
463	"before-extauth-disable": before_extauth_disable,
464	}
465	XenAPIPlugin.dispatch(dispatch_tbl)	×

xapi-project / xen-api / 12583462520

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous