22

✏️ Editing: authcompat.py

#!/usr/libexec/platform-python
# -*- coding: utf-8 -*-
#
#    Authors:
#        Pavel Březina <pbrezina@redhat.com>
#
#    Copyright (C) 2018 Red Hat
#
#    This program is free software; you can redistribute it and/or modify
#    it under the terms of the GNU General Public License as published by
#    the Free Software Foundation; either version 3 of the License, or
#    (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU General Public License for more details.
#
#    You should have received a copy of the GNU General Public License
#    along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

import os
import sys
import locale
import gettext
import subprocess

from authcompat_Options import Options
from authcompat_EnvironmentFile import EnvironmentFile
from authcompat_ConfigSnippet import ConfigSnippet

_ = gettext.gettext

def eprint(*args, **kwargs):
    print(*args, file=sys.stderr, **kwargs)

class Command:
    TEST = False

def __init__(self, command, args, input=None, check=True):
        self.args = [command] + args
        self.input = input.encode() if input is not None else None
        self.check = check
        self.result = None

def run(self):
        print(_("Executing: %s") % ' '.join(self.args))

if self.TEST:
            return

self.result = subprocess.run(self.args, check=self.check,
                                     input=self.input,
                                     stdout=subprocess.PIPE,
                                     stderr=subprocess.PIPE)

class Service:
    def __init__(self, name):
        self.name = name + '.service'

def runsystemd(self, command, required, enoent_code):
        try:
            command.run()
        except subprocess.CalledProcessError as result:
            if required and result.returncode == enoent_code:
                eprint(_("Service %s was not found. Please install the service.")
                       % self.name)
            elif result.returncode != enoent_code:
                eprint(_("Command [%s] failed with %d, stderr:")
                       % (' '.join(result.cmd), result.returncode))
                eprint(result.stderr.decode())

def enable(self):
        cmd = Command(Path.System("cmd-systemctl"), ["enable", self.name])
        self.runsystemd(cmd, True, 1)

def disable(self):
        cmd = Command(Path.System("cmd-systemctl"), ["disable", self.name])
        self.runsystemd(cmd, False, 1)

def start(self, Restart=True):
        if Restart:
            self.stop()
        cmd = Command(Path.System("cmd-systemctl"), ["start", self.name])
        self.runsystemd(cmd, True, 5)

def stop(self):
        cmd = Command(Path.System("cmd-systemctl"), ["stop", self.name])
        self.runsystemd(cmd, False, 5)

class Path:
    LocalDir = os.path.dirname(os.path.realpath(__file__))
    Config = EnvironmentFile(LocalDir + "/authcompat_paths")

Files = {
        'ldap.conf': '/etc/openldap/ldap.conf',
        'krb5.conf': '/etc/krb5.conf.d/authconfig-krb.conf',
        'sssd.conf': '/etc/sssd/conf.d/authconfig-sssd.conf',
        'authconfig': '/etc/sysconfig/authconfig',
        'network': '/etc/sysconfig/network',
        'pwquality.conf': '/etc/security/pwquality.conf.d/10-authconfig-pwquality.conf',
        'yp.conf': '/etc/yp.conf',
        'cmd-systemctl': '/usr/bin/systemctl',
        'cmd-authselect': '/usr/bin/authselect',
        'cmd-realm': '/usr/sbin/realm',
        'cmd-domainname': '/usr/bin/domainname',
        'cmd-setsebool': '/usr/sbin/setsebool'
    }

@staticmethod
    def Local(relpath):
        return "%s/%s" % (Path.LocalDir, relpath)

@staticmethod
    def System(name):
        return Path.Files[name]

class Configuration:
    class Base(object):
        def __init__(self, options, ServiceName=None):
            self.options = options
            self.service = None
            if ServiceName is not None:
                self.service = Service(ServiceName)

def isEnabled(self):
            return True

def isDisabled(self):
            return not self.isEnabled()

def enableService(self, nostart):
            if self.service is None:
                return

self.service.enable()

if not nostart:
                self.service.start()

def disableService(self, nostop):
            if self.service is None:
                return

self.service.disable()

if not nostop:
                self.service.stop()

def cleanup(self):
            return

def write(self):
            return

def get(self, name):
            return self.options.get(name)

def isset(self, name):
            return self.options.isset(name)

def getTrueOrNone(self, name):
            return self.options.getTrueOrNone(name)

def getBool(self, name):
            return self.options.getBool(name)

def getBoolAsValue(self, name, if_true, if_false, AllowNone=False):
            if AllowNone and not self.isset(name):
                return None

value = self.getBool(name)
            if value:
                return if_true

return if_false

def removeFile(self, filename):
            print(_("Removing file: %s") % filename)
            if self.options.getBool("test-call"):
                return

try:
                os.remove(filename)
            except FileNotFoundError:
                return

class LDAP(Base):
        def __init__(self, options):
            super(Configuration.LDAP, self).__init__(options)

def write(self):
            config = EnvironmentFile(Path.System('ldap.conf'), " ",
                                     delimiter_re=r"\s\t", quotes=False)
            if self.isset("ldapserver"):
                config.set("URI", self.get("ldapserver"))

if self.isset("ldapbasedn"):
                config.set("BASE", self.get("ldapbasedn"))

config.write()

class Kerberos(Base):
        def __init__(self, options):
            super(Configuration.Kerberos, self).__init__(options)

def isEnabled(self):
            if not self.isset("krb5realm") and not self.isset("krb5realmdns"):
                return None

return self.get("krb5realm") != "" or self.getBool("krb5realmdns")

def cleanup(self):
            # Do not remove the file if these options are not set
            if not self.isset("krb5realm") and not self.isset("krb5realmdns"):
                return

self.removeFile(Path.System('krb5.conf'))

def write(self):
            if self.isDisabled():
                return

path = Path.Local("snippets/authconfig-krb.conf")
            config = ConfigSnippet(path, Path.System('krb5.conf'))
            realm = self.get("krb5realm")

keys = {
                'realm': self.get("krb5realm"),
                'kdc-srv': self.get("krb5kdcdns"),
                'realm-srv': self.get("krb5realmdns"),
                'kdc': self.get("krb5kdc") if realm else None,
                'adminserver': self.get("krb5adminserver") if realm else None,
                'domain': realm.lower() if realm else None
            }

config.write(keys)

class Network(Base):
        def __init__(self, options):
            super(Configuration.Network, self).__init__(options)

def write(self):
            nisdomain = self.get("nisdomain")
            config = EnvironmentFile(Path.System('network'))

if nisdomain is None:
                return

config.set("NISDOMAIN", nisdomain)
            config.write()

class SSSD(Base):
        def __init__(self, options):
            super(Configuration.SSSD, self).__init__(options, ServiceName="sssd")

def isEnabled(self):
            if not self.isset("ldap") and not self.isset("sssd"):
                return None

return self.getBool("ldap") or self.getBool("sssd")

def cleanup(self):
            self.removeFile(Path.System('sssd.conf'))

def write(self):
            # Authconfig would not generate sssd in this case so we should not
            # either. Even if --enablesssd[auth] was provided the configuration
            # would not be generated.
            if not self.getBool("ldap"):
                return

path = Path.Local("snippets/authconfig-sssd.conf")
            config = ConfigSnippet(path, Path.System('sssd.conf'))

schema = "rfc2307bis" if self.getBool("rfc2307bis") else None

keys = {
                'ldap-uri': self.get("ldapserver"),
                'ldap-basedn': self.get("ldapbasedn"),
                'ldap-tls': self.getTrueOrNone("ldaptls"),
                'ldap-schema': schema,
                'krb5': self.getTrueOrNone("krb5"),
                'kdc-uri': self.get("krb5kdc"),
                'kpasswd-uri': self.get("krb5adminserver"),
                'realm': self.get("krb5realm"),
                'cache-creds': self.getTrueOrNone("cachecreds"),
                'cert-auth': self.getTrueOrNone("smartcard")
            }

config.write(keys)
            os.chmod(Path.System('sssd.conf'), mode=0o600)

class Winbind(Base):
        def __init__(self, options):
            super(Configuration.Winbind, self).__init__(options, ServiceName="winbind")

def isEnabled(self):
            if not self.isset("winbind") and not self.isset("winbindauth"):
                return None

return self.getBool("winbind") or self.getBool("winbindauth")

def write(self):
            if not self.isset("winbindjoin"):
                return

creds = self.options.get("winbindjoin").split("%", 1)

user = creds[0]
            password = None
            if len(creds) > 1:
                password = creds[1] + '\n'

args = [
                'join',
                '-U', '"%s"' % user,
                '--client-software', 'winbind'
            ]

if self.isset("smbworkgroup"):
                args.append(self.get("smbworkgroup"))

cmd = Command(Path.System('cmd-realm'), args, input=password)
            try:
                cmd.run()
            except FileNotFoundError:
                eprint(_("%s was not found. Please, install realmd.")
                       % Path.System('cmd-realm'))

class PWQuality(Base):
        def __init__(self, options):
            super(Configuration.PWQuality, self).__init__(options)

def write(self):
            config = EnvironmentFile(Path.System('pwquality.conf'))
            value_set = False

pwopts = {
                "minlen": self.get("passminlen"),
                "minclass": self.get("passminclass"),
                "maxrepeat": self.get("passmaxrepeat"),
                "maxclassrepeat": self.get("passmaxclassrepeat"),
                "lcredit": self.getBoolAsValue("reqlower", -1, 0, AllowNone=True),
                "ucredit": self.getBoolAsValue("requpper", -1, 0, AllowNone=True),
                "dcredit": self.getBoolAsValue("reqdigit", -1, 0, AllowNone=True),
                "ocredit": self.getBoolAsValue("reqother", -1, 0, AllowNone=True)
            }

# Write options only if their are actually set
            for opt, value in pwopts.items():
                if value is not None:
                    print(opt + "=" + str(value))
                    config.set(opt, value)
                    value_set = True

if value_set:
                config.write()

class MakeHomedir(Base):
        def __init__(self, options):
            super(Configuration.MakeHomedir, self).__init__(options, ServiceName="oddjobd")

def isEnabled(self):
            if not self.isset("mkhomedir"):
                return None

return self.getBool("mkhomedir")

def disableService(self, nostop):
            # Never disable the service in case it is already running as
            # other applications may depend on it.
            return

class NIS(Base):
        def __init__(self, options):
            super(Configuration.NIS, self).__init__(options)
            self.rpcbind = Service("rpcbind")
            self.ypbind = Service("ypbind")

def isEnabled(self):
            if not self.isset("nis"):
                return None

return self.getBool("nis")

def enableService(self, nostart):
            if not self.isset("nisdomain"):
                return

nisdom = self.get("nisdomain")

if not nostart:
                cmd = Command(Path.System('cmd-domainname'), [nisdom])
                cmd.run()

cmd = Command(Path.System('cmd-setsebool'),
                          ['-P', 'allow_ypbind', '1'])
            cmd.run()

self.rpcbind.enable()
            self.ypbind.enable()

if not nostart:
                self.rpcbind.start(Restart=False)
                self.ypbind.start()

def disableService(self, nostop):
            if not nostop:
                cmd = Command(Path.System('cmd-domainname'), ["(none)"])
                cmd.run()

cmd = Command(Path.System('cmd-setsebool'),
                          ['-P', 'allow_ypbind', '0'])
            cmd.run()

self.rpcbind.disable()
            self.ypbind.disable()

if not nostop:
                self.rpcbind.stop()
                self.ypbind.stop()

def write(self):
            if not self.isset("nisdomain"):
                return

output = "domain " + self.get("nisdomain")

additional_servers = []
            if self.isset("nisserver"):
                servers = self.get("nisserver").split(",")
                additional_servers = servers[1:]
                output += " server " + servers[0] + "\n"
            else:
                output += " broadcast\n"

for server in additional_servers:
                output += "ypserver " + server + "\n"

filename = Path.System('yp.conf')
            if self.getBool("test-call"):
                print("========== BEGIN Content of [%s] ==========" % filename)
                print(output)
                print("========== END   Content of [%s] ==========\n" % filename)
                return

with open(filename, "w") as f:
                f.write(output)

class AuthCompat:
    def __init__(self):
        self.sysconfig = EnvironmentFile(Path.System('authconfig'))
        self.options = Options()

self.options.parse()
        self.options.applysysconfig(self.sysconfig)
        self.options.updatesysconfig(self.sysconfig)

def printWarning(self):
        print(_("Running authconfig compatibility tool."))
        print(_("The purpose of this tool is to enable authentication against "
                "chosen services with authselect and minimum configuration. "
                "It does not provide all capabilities of authconfig.\n"))
        print(_("IMPORTANT: authconfig is replaced by authselect, "
                "please update your scripts."))
        print(_("See man authselect-migration(7) to help you with migration to authselect"))

options = self.options.getSetButUnsupported()
        if options:
            print(_("Warning: These options are not supported anymore "
                    "and have no effect:"))
            for name in options:
                print("  --%s" % name)

print("")

def printOptions(self):
        for option in Options.List:
            print("%s=%s" % (option.name, option.value))

def printSysconfig(self):
        for line in self.sysconfig.getall():
            print("%s=%s" % (line.name, line.value))

def canContinue(self):
        disallowed = ["test", "probe", "restorebackup", "restorelastbackup"]
        required = ["update", "updateall", "kickstart"]

if not self.options.getBool("test") and os.getuid() != 0:
            print(_("authconfig can only be run as root"))
            return False

for option in disallowed:
            if self.options.getBool(option):
                print(_("Error: option --%s is no longer supported and we "
                        "cannot continue if it is set." % option))
                return False

if self.options.getBool("winbind") != self.options.getBool("winbindauth"):
            print(_("Error: Both --enablewinbind and --enablewinbindauth must be set."))
            return False

# We require one of these options to perform changes
        # We encourage to use --updateall since we no longer support just pure
        # --update or --kickstart, they will act as --updateall.
        for option in required:
            if self.options.getBool(option):
                return True

print(_("Error: Please, provide --updateall option."))
        return False

def runAuthselect(self):
        map = {
            'smartcard': 'with-smartcard',
            'requiresmartcard': 'with-smartcard-required',
            'fingerprint': 'with-fingerprint',
            'mkhomedir': 'with-mkhomedir',
            'faillock': 'with-faillock',
            'pamaccess': 'with-pamaccess',
            'winbindkrb5': 'with-krb5'
        }

# Read current configuration first.
        (profile, features) = self.getCurrentAuthselectConfig()

# Change profile if requested.
        if (self.options.getBool("ldap")
                or self.options.getBool("ldapauth")
                or self.options.getBool("sssd")
                or self.options.getBool("sssdauth")):
            profile = "sssd"
        elif self.options.getBool("nis"):
            profile = "nis"
        elif self.options.getBool("winbind"):
            profile = "winbind"

# Default to sssd
        if profile is None:
            profile = "sssd"

# Add enabled and remove disabled features.
        for option, feature in map.items():
            if not self.options.isset(option):
                continue

enabled = self.options.getBool(option)
            if enabled:
                features.append(feature)
            else:
                while feature in features:
                    features.remove(feature)

# Add lock-on-smartcard-removal if requested
        if self.options.isset("smartcardaction"):
            if int(self.options.get("smartcardaction")) == 0:
                features.append("with-smartcard-lock-on-removal")
            else:
                features.remove("with-smartcard-lock-on-removal")

# Remove duplicates. The order is not kept but that does not matter.
        features = list(set(features))

# Always run with --force. This is either first call of authconfig
        # in installation script or it is run on already configured system.
        # We want to use authselect in both cases anyway, since authconfig
        # would change the configuration either way.
        args = ["select", profile]
        args.extend(features)
        args.append("--force")

cmd = Command(Path.System('cmd-authselect'), args)
        cmd.run()

def getCurrentAuthselectConfig(self):
        cmd = Command(Path.System('cmd-authselect'), ['check'], check=False)
        cmd.run()

if cmd.result is None or cmd.result.returncode != 0:
            return (None, [])

cmd = Command(Path.System('cmd-authselect'), ['current', '--raw'])
        cmd.run()

current = cmd.result.stdout.decode("utf-8").split()
        return (current[0], current[1:])

def writeConfiguration(self):
        configs = [
            Configuration.LDAP(self.options),
            Configuration.Network(self.options),
            Configuration.Kerberos(self.options),
            Configuration.SSSD(self.options),
            Configuration.Winbind(self.options),
            Configuration.PWQuality(self.options),
            Configuration.MakeHomedir(self.options),
            Configuration.NIS(self.options)
        ]

for config in configs:
            # Configuration decides if it needs to write something or not
            config.write()

# Enable or disable service if needed
            nostart = self.options.getBool("nostart")
            try:
                enabled = config.isEnabled()

# Skip service management if it can not be decided
                if enabled is None:
                    continue

if enabled:
                    config.enableService(nostart)
                else:
                    config.disableService(nostart)
                    config.cleanup()
            except subprocess.CalledProcessError as result:
                # This is not fatal error.
                eprint(_("Command [%s] failed with %d, stderr:")
                       % (' '.join(result.cmd), result.returncode))
                eprint(result.stderr.decode())

def main():
    try:
        locale.setlocale(locale.LC_ALL, '')
    except locale.Error:
        sys.stderr.write('Warning: Unsupported locale setting.\n')

authcompat = AuthCompat()
    authcompat.printWarning()

Command.TEST = authcompat.options.getBool("test-call")
    EnvironmentFile.TEST = authcompat.options.getBool("test-call")
    ConfigSnippet.TEST = authcompat.options.getBool("test-call")

if not authcompat.canContinue():
        sys.exit(1)

try:
        authcompat.runAuthselect()
        authcompat.writeConfiguration()
        authcompat.sysconfig.write()
    except subprocess.CalledProcessError as result:
        eprint(_("Command [%s] failed with %d, stderr:")
               % (' '.join(result.cmd), result.returncode))
        eprint(result.stderr.decode())

sys.exit(0)

if __name__ == "__main__":
    main()

MR SKB