Get MySQL hardening config audits.

    :returns:  dictionary of audits
189610
def get_audits():
    
    if subprocess.call(['which', 'mysql'], stdout=subprocess.PIPE) != 0:
        log("MySQL does not appear to be installed on this node - "
            "skipping mysql hardening", level=WARNING)
        return []

    settings = utils.get_settings('mysql')
    hardening_settings = settings['hardening']
    my_cnf = hardening_settings['mysql-conf']

    audits = [
        FilePermissionAudit(paths=[my_cnf], user='root',
                            group='root', mode=0o0600),

        TemplatedFile(hardening_settings['hardening-conf'],
                      MySQLConfContext(),
                      TEMPLATES_DIR,
                      mode=0o0750,
                      user='mysql',
                      group='root',
                      service_actions=[{'service': 'mysql',
                                        'actions': ['restart']}]),

        # MySQL and Percona charms do not allow configuration of the
        # data directory, so use the default.
        DirectoryPermissionAudit('/var/lib/mysql',
                                 user='mysql',
                                 group='mysql',
                                 recursive=False,
                                 mode=0o755),

        DirectoryPermissionAudit('/etc/mysql',
                                 user='root',
                                 group='root',
                                 recursive=False,
                                 mode=0o700),
    ]

    return audits
1360432
def get_configuration(self):
        
        mapping = {}
        settings = self.get_settings()
        for record in self.context.getAnalyses():
            uid = record.get("service_uid")
            setting = settings.get(uid, {})
            config = {
                "partition": record.get("partition"),
                "hidden": setting.get("hidden", False),
            }
            mapping[uid] = config
        return mapping
313415
def _get_generator_plugins(cls):
        

        if not hasattr(cls, 'suite_generators'):
            gens = working_set.iter_entry_points('compliance_checker.generators')
            cls.suite_generators = [x.resolve() for x in gens]

        return cls.suite_generators
397813
def get_audits():
    
    audits = []
    settings = utils.get_settings('os')

    # Ensure that the /etc/security/limits.d directory is only writable
    # by the root user, but others can execute and read.
    audits.append(DirectoryPermissionAudit('/etc/security/limits.d',
                                           user='root', group='root',
                                           mode=0o755))

    # If core dumps are not enabled, then don't allow core dumps to be
    # created as they may contain sensitive information.
    if not settings['security']['kernel_enable_core_dump']:
        audits.append(TemplatedFile('/etc/security/limits.d/10.hardcore.conf',
                                    SecurityLimitsContext(),
                                    template_dir=TEMPLATES_DIR,
                                    user='root', group='root', mode=0o0440))
    return audits
1240460
def get_audits():
    
    audits = []
    settings = utils.get_settings('os')

    # Remove write permissions from $PATH folders for all regular users.
    # This prevents changing system-wide commands from normal users.
    path_folders = {'/usr/local/sbin',
                    '/usr/local/bin',
                    '/usr/sbin',
                    '/usr/bin',
                    '/bin'}
    extra_user_paths = settings['environment']['extra_user_paths']
    path_folders.update(extra_user_paths)
    audits.append(ReadOnly(path_folders))

    # Only allow the root user to have access to the shadow file.
    audits.append(FilePermissionAudit('/etc/shadow', 'root', 'root', 0o0600))

    if 'change_user' not in settings['security']['users_allow']:
        # su should only be accessible to user and group root, unless it is
        # expressly defined to allow users to change to root via the
        # security_users_allow config option.
        audits.append(FilePermissionAudit('/bin/su', 'root', 'root', 0o750))

    return audits
400299
def accounts(self):
        
        d = {}

        if False and not self._account_password:
            from ambry.dbexceptions import ConfigurationError
            raise ConfigurationError(
                "Can't access accounts without setting an account password"
                " either in the accounts.password config, or in the AMBRY_ACCOUNT_PASSWORD"
                " env var.")

        for act in self.database.session.query(Account).all():
            if self._account_password:
                act.secret_password = self._account_password
            e = act.dict
            a_id = e['account_id']
            d[a_id] = e

        return d
413610
def mysql_to_dict(data, key):
    '''
    
    '''
    ret = {}
    headers = ['']
    for line in data:
        if not line:
            continue
        if line.startswith('+'):
            continue
        comps = line.split('|')
        for comp in range(len(comps)):
            comps[comp] = comps[comp].strip()
        if len(headers) > 1:
            index = len(headers) - 1
            row = {}
            for field in range(index):
                if field < 1:
                    continue
                else:
                    row[headers[field]] = salt.utils.stringutils.to_num(comps[field])
            ret[row[key]] = row
        else:
            headers = comps
    return ret
791659
def _get_options(self):
        '''
        
        '''
        defaults = {'host': 'localhost',
                    'user': 'salt',
                    'pass': 'salt',
                    'db': 'salt',
                    'port': 3306,
                    'ssl': {}}
        _options = {}
        _opts = __opts__.get('mysql', {})
        for attr in defaults:
            if attr not in _opts:
                log.debug('Using default for MySQL %s', attr)
                _options[attr] = defaults[attr]
                continue
            _options[attr] = _opts[attr]
        return _options
937919
def get_params_from_sqlalchemy_url(db_url):
        # type: (str) -> Dict[str,Any]
        
        result = urlsplit(db_url)
        return {'database': result.path[1:], 'host': result.hostname, 'port': result.port,
                'username': result.username, 'password': result.password, 'driver': result.scheme}
6202
def to_dict(self):
        """"""
        db_dict = self.credentials
        db_dict.update(self.tables.to_dict())
        return db_dict
380606
def get_mysql_connection(host, user, port, password, database, ssl={}):
    

    return pymysql.connect(host=host,
                           user=user,
                           port=port,
                           password=password,
                           db=database,
                           charset='utf8mb4',
                           cursorclass=pymysql.cursors.DictCursor,
                           client_flag=pymysql.constants.CLIENT.MULTI_STATEMENTS,
                           ssl=ssl
                           )
1154693
def get_config(self):
        
        # Fill this dict with config data
        return {
            'hash_name': self.hash_name,
            'dim': self.dim,
            'bin_width': self.bin_width,
            'projection_count': self.projection_count,
            'normals': self.normals
        }
492593
def defaults(self):
        
        defaults = ConfigParser.defaults(self).copy()
        for key, val in iteritems(defaults):
            defaults[key] = self.get('DEFAULT', key) or val
        return defaults
665774
def settings_to_cmd_args(settings_dict):
    
    args = ['mysql']
    db = settings_dict['OPTIONS'].get('db', settings_dict['NAME'])
    user = settings_dict['OPTIONS'].get('user', settings_dict['USER'])
    passwd = settings_dict['OPTIONS'].get('passwd', settings_dict['PASSWORD'])
    host = settings_dict['OPTIONS'].get('host', settings_dict['HOST'])
    port = settings_dict['OPTIONS'].get('port', settings_dict['PORT'])
    cert = settings_dict['OPTIONS'].get('ssl', {}).get('ca')
    defaults_file = settings_dict['OPTIONS'].get('read_default_file')
    # Seems to be no good way to set sql_mode with CLI.

    if defaults_file:
        args += ["--defaults-file=%s" % defaults_file]
    if user:
        args += ["--user=%s" % user]
    if passwd:
        args += ["--password=%s" % passwd]
    if host:
        if '/' in host:
            args += ["--socket=%s" % host]
        else:
            args += ["--host=%s" % host]
    if port:
        args += ["--port=%s" % port]
    if cert:
        args += ["--ssl-ca=%s" % cert]
    if db:
        args += [db]
    return args
955030
def to_dict(self):
        
        return dict(
            host=self.host,
            port=self.port,
            database=self.database,
            username=self.username,
            password=self.password,
        )
1275775
def get_protected_settings(self, settings_source):
        
        user_backend = settings_source.get('keyring_backend')
        found_settings = dict()
        for setting_name in self.__PROTECTED_SETTINGS:
            # Nothing to do if the setting is already defined.
            if settings_source.get(setting_name):
                continue

            setting = None
            identifier = "%s.%s" % (self.__name__, setting_name)
            if hasattr(self, 'required') and setting_name in getattr(self, 'required'):
                setting = self.get_setting_from_keyring(identifier, user_backend)
            elif hasattr(self, setting_name):
                setting = self.get_setting_from_keyring(identifier, user_backend)
            if setting:
                found_settings.update({setting_name: setting})
        return found_settings
287780
def get_mysql_vars(mysql: str,
                   host: str,
                   port: int,
                   user: str) -> Dict[str, str]:
    
    cmdargs = [
        mysql,
        "-h", host,
        "-P", str(port),
        "-e", "SHOW VARIABLES; SHOW STATUS",
        "-u", user,
        "-p"  # prompt for password
    ]
    log.info("Connecting to MySQL with user: {}", user)
    log.debug(cmdargs)
    process = subprocess.Popen(cmdargs, stdout=subprocess.PIPE)
    out, err = process.communicate()
    lines = out.decode("utf8").splitlines()
    mysqlvars = {}
    for line in lines:
        var, val = line.split("\t")
        mysqlvars[var] = val
    return mysqlvars