Skip to content

Commit

Permalink
fix ansible-linting after rebase onto main
Browse files Browse the repository at this point in the history
  • Loading branch information
@youtous
youtous committed Dec 1, 2023
1 parent d5093fa commit ec8e04c
Show file tree
Hide file tree
Showing 29 changed files with 257 additions and 198 deletions.
50 changes: 27 additions & 23 deletions collections/ansible_collections/demo/patching/plugins/modules/scan_packages.py
View file
Original file line number Diff line number Diff line change
@@ -1,16 +1,16 @@
#!/usr/bin/env python

from ansible.module_utils.basic import * # noqa
from ansible.module_utils.basic import * # noqa

DOCUMENTATION = '''
DOCUMENTATION = """
---
module: scan_packages
short_description: Return installed packages information as fact data
description:
- Return information about installed packages as fact data
'''
"""

EXAMPLES = '''
EXAMPLES = """
# Example fact output:
# host | success >> {
# "ansible_facts": {
Expand All @@ -34,21 +34,23 @@
# "name": "gcc-4.8-base"
# }
# ]
'''
"""


def rpm_package_list():
import rpm

trans_set = rpm.TransactionSet()
installed_packages = []
for package in trans_set.dbMatch():
package_details = {
'name':package[rpm.RPMTAG_NAME],
'version':package[rpm.RPMTAG_VERSION],
'release':package[rpm.RPMTAG_RELEASE],
'epoch':package[rpm.RPMTAG_EPOCH],
'arch':package[rpm.RPMTAG_ARCH],
'source':'rpm' }
"name": package[rpm.RPMTAG_NAME],
"version": package[rpm.RPMTAG_VERSION],
"release": package[rpm.RPMTAG_RELEASE],
"epoch": package[rpm.RPMTAG_EPOCH],
"arch": package[rpm.RPMTAG_ARCH],
"source": "rpm",
}
if installed_packages == []:
installed_packages = [package_details]
else:
Expand All @@ -58,16 +60,20 @@ def rpm_package_list():

def deb_package_list():
import apt

apt_cache = apt.Cache()
installed_packages = []
apt_installed_packages = [pk for pk in apt_cache.keys() if apt_cache[pk].is_installed]
apt_installed_packages = [
pk for pk in apt_cache.keys() if apt_cache[pk].is_installed
]
for package in apt_installed_packages:
ac_pkg = apt_cache[package].installed
package_details = {
'name':package,
'version':ac_pkg.version,
'arch':ac_pkg.architecture,
'source':'apt'}
"name": package,
"version": ac_pkg.version,
"arch": ac_pkg.architecture,
"source": "apt",
}
if installed_packages == []:
installed_packages = [package_details]
else:
Expand All @@ -76,13 +82,11 @@ def deb_package_list():


def main():
module = AnsibleModule(
argument_spec = dict(os_family=dict(required=True))
)
ans_os = module.params['os_family']
if ans_os in ('RedHat', 'Suse', 'openSUSE Leap'):
module = AnsibleModule(argument_spec=dict(os_family=dict(required=True)))
ans_os = module.params["os_family"]
if ans_os in ("RedHat", "Suse", "openSUSE Leap"):
packages = rpm_package_list()
elif ans_os == 'Debian':
elif ans_os == "Debian":
packages = deb_package_list()
else:
packages = None
Expand All @@ -94,4 +98,4 @@ def main():
module.exit_json(**results)


main()
main()
169 changes: 108 additions & 61 deletions collections/ansible_collections/demo/patching/plugins/modules/scan_services.py
View file
Original file line number Diff line number Diff line change
@@ -1,46 +1,47 @@
#!/usr/bin/env python

import re
from ansible.module_utils.basic import * # noqa
from ansible.module_utils.basic import * # noqa

DOCUMENTATION = '''
DOCUMENTATION = """
---
module: scan_services
short_description: Return service state information as fact data
description:
- Return service state information as fact data for various service management utilities
'''
"""

EXAMPLES = '''
EXAMPLES = """
---
- monit: scan_services
# Example fact output:
# host | success >> {

Check failure on line 19 in collections/ansible_collections/demo/patching/plugins/modules/scan_services.py

View workflow job for this annotation

GitHub Actions / Ansible Lint

yaml[comments] 

Missing starting space in comment
# "ansible_facts": {
# "services": {
# "network": {
# "source": "sysv",
# "state": "running",
# "name": "network"
# },
# "arp-ethers.service": {
# "source": "systemd",
# "state": "stopped",
# "name": "arp-ethers.service"
# }
# }
# "ansible_facts": {
# "services": {
# "network": {
# "source": "sysv",
# "state": "running",
# "name": "network"
# },
# "arp-ethers.service": {
# "source": "systemd",
# "state": "stopped",
# "name": "arp-ethers.service"

Check failure on line 30 in collections/ansible_collections/demo/patching/plugins/modules/scan_services.py

View workflow job for this annotation

GitHub Actions / Ansible Lint

yaml[comments] 

Missing starting space in comment
# }
# }
# }
'''
# }
"""


class BaseService(object):

def __init__(self, module):
self.module = module
self.incomplete_warning = False


class ServiceScanService(BaseService):

def gather_services(self):
services = {}
service_path = self.module.get_bin_path("service")
Expand All @@ -51,105 +52,142 @@ def gather_services(self):

# sysvinit
if service_path is not None and chkconfig_path is None:
rc, stdout, stderr = self.module.run_command("%s --status-all 2>&1 | grep -E \"\\[ (\\+|\\-) \\]\"" % service_path, use_unsafe_shell=True)
rc, stdout, stderr = self.module.run_command(
'%s --status-all 2>&1 | grep -E "\\[ (\\+|\\-) \\]"' % service_path,
use_unsafe_shell=True,
)
for line in stdout.split("\n"):
line_data = line.split()
if len(line_data) < 4:
continue # Skipping because we expected more data
continue # Skipping because we expected more data
service_name = " ".join(line_data[3:])
if line_data[1] == "+":
service_state = "running"
else:
service_state = "stopped"
services[service_name] = {"name": service_name, "state": service_state, "source": "sysv"}
services[service_name] = {
"name": service_name,
"state": service_state,
"source": "sysv",
}

# Upstart
if initctl_path is not None and chkconfig_path is None:
p = re.compile('^\s?(?P<name>.*)\s(?P<goal>\w+)\/(?P<state>\w+)(\,\sprocess\s(?P<pid>[0-9]+))?\s*$')
p = re.compile(
"^\s?(?P<name>.*)\s(?P<goal>\w+)\/(?P<state>\w+)(\,\sprocess\s(?P<pid>[0-9]+))?\s*$"
)
rc, stdout, stderr = self.module.run_command("%s list" % initctl_path)
real_stdout = stdout.replace("\r","")
real_stdout = stdout.replace("\r", "")
for line in real_stdout.split("\n"):
m = p.match(line)
if not m:
continue
service_name = m.group('name')
service_goal = m.group('goal')
service_state = m.group('state')
if m.group('pid'):
pid = m.group('pid')
service_name = m.group("name")
service_goal = m.group("goal")
service_state = m.group("state")
if m.group("pid"):
pid = m.group("pid")
else:
pid = None # NOQA
payload = {"name": service_name, "state": service_state, "goal": service_goal, "source": "upstart"}
payload = {
"name": service_name,
"state": service_state,
"goal": service_goal,
"source": "upstart",
}
services[service_name] = payload

# RH sysvinit
elif chkconfig_path is not None:
#print '%s --status-all | grep -E "is (running|stopped)"' % service_path
# print '%s --status-all | grep -E "is (running|stopped)"' % service_path
p = re.compile(
'(?P<service>.*?)\s+[0-9]:(?P<rl0>on|off)\s+[0-9]:(?P<rl1>on|off)\s+[0-9]:(?P<rl2>on|off)\s+'
'[0-9]:(?P<rl3>on|off)\s+[0-9]:(?P<rl4>on|off)\s+[0-9]:(?P<rl5>on|off)\s+[0-9]:(?P<rl6>on|off)')
rc, stdout, stderr = self.module.run_command('%s' % chkconfig_path, use_unsafe_shell=True)
"(?P<service>.*?)\s+[0-9]:(?P<rl0>on|off)\s+[0-9]:(?P<rl1>on|off)\s+[0-9]:(?P<rl2>on|off)\s+"
"[0-9]:(?P<rl3>on|off)\s+[0-9]:(?P<rl4>on|off)\s+[0-9]:(?P<rl5>on|off)\s+[0-9]:(?P<rl6>on|off)"
)
rc, stdout, stderr = self.module.run_command(
"%s" % chkconfig_path, use_unsafe_shell=True
)
# Check for special cases where stdout does not fit pattern
match_any = False
for line in stdout.split('\n'):
for line in stdout.split("\n"):
if p.match(line):
match_any = True
if not match_any:
p_simple = re.compile('(?P<service>.*?)\s+(?P<rl0>on|off)')
p_simple = re.compile("(?P<service>.*?)\s+(?P<rl0>on|off)")
match_any = False
for line in stdout.split('\n'):
for line in stdout.split("\n"):
if p_simple.match(line):
match_any = True
if match_any:
# Try extra flags " -l --allservices" needed for SLES11
rc, stdout, stderr = self.module.run_command('%s -l --allservices' % chkconfig_path, use_unsafe_shell=True)
elif '--list' in stderr:
rc, stdout, stderr = self.module.run_command(
"%s -l --allservices" % chkconfig_path, use_unsafe_shell=True
)
elif "--list" in stderr:
# Extra flag needed for RHEL5
rc, stdout, stderr = self.module.run_command('%s --list' % chkconfig_path, use_unsafe_shell=True)
for line in stdout.split('\n'):
rc, stdout, stderr = self.module.run_command(
"%s --list" % chkconfig_path, use_unsafe_shell=True
)
for line in stdout.split("\n"):
m = p.match(line)
if m:
service_name = m.group('service')
service_state = 'stopped'
if m.group('rl3') == 'on':
rc, stdout, stderr = self.module.run_command('%s %s status' % (service_path, service_name), use_unsafe_shell=True)
service_name = m.group("service")
service_state = "stopped"
if m.group("rl3") == "on":
rc, stdout, stderr = self.module.run_command(
"%s %s status" % (service_path, service_name),
use_unsafe_shell=True,
)
service_state = rc
if rc in (0,):
service_state = 'running'
#elif rc in (1,3):
service_state = "running"
# elif rc in (1,3):
else:
if 'root' in stderr or 'permission' in stderr.lower() or 'not in sudoers' in stderr.lower():
if (
"root" in stderr
or "permission" in stderr.lower()
or "not in sudoers" in stderr.lower()
):
self.incomplete_warning = True
continue
else:
service_state = 'stopped'
service_data = {"name": service_name, "state": service_state, "source": "sysv"}
service_state = "stopped"
service_data = {
"name": service_name,
"state": service_state,
"source": "sysv",
}
services[service_name] = service_data
return services


class SystemctlScanService(BaseService):

def systemd_enabled(self):
# Check if init is the systemd command, using comm as cmdline could be symlink
try:
f = open('/proc/1/comm', 'r')
f = open("/proc/1/comm", "r")
except IOError:
# If comm doesn't exist, old kernel, no systemd
return False
for line in f:
if 'systemd' in line:
if "systemd" in line:
return True
return False

def gather_services(self):
services = {}
if not self.systemd_enabled():
return None
systemctl_path = self.module.get_bin_path("systemctl", opt_dirs=["/usr/bin", "/usr/local/bin"])
systemctl_path = self.module.get_bin_path(
"systemctl", opt_dirs=["/usr/bin", "/usr/local/bin"]
)
if systemctl_path is None:
return None
rc, stdout, stderr = self.module.run_command("%s list-unit-files --type=service | tail -n +2 | head -n -2" % systemctl_path, use_unsafe_shell=True)
rc, stdout, stderr = self.module.run_command(
"%s list-unit-files --type=service | tail -n +2 | head -n -2"
% systemctl_path,
use_unsafe_shell=True,
)
for line in stdout.split("\n"):
line_data = line.split()
if len(line_data) != 2:
Expand All @@ -158,12 +196,16 @@ def gather_services(self):
state_val = "running"
else:
state_val = "stopped"
services[line_data[0]] = {"name": line_data[0], "state": state_val, "source": "systemd"}
services[line_data[0]] = {
"name": line_data[0],
"state": state_val,
"source": "systemd",
}
return services


def main():
module = AnsibleModule(argument_spec = dict())
module = AnsibleModule(argument_spec=dict())
service_modules = (ServiceScanService, SystemctlScanService)
all_services = {}
incomplete_warning = False
Expand All @@ -175,12 +217,17 @@ def main():
if svcmod.incomplete_warning:
incomplete_warning = True
if len(all_services) == 0:
results = dict(skipped=True, msg="Failed to find any services. Sometimes this is due to insufficient privileges.")
results = dict(
skipped=True,
msg="Failed to find any services. Sometimes this is due to insufficient privileges.",
)
else:
results = dict(ansible_facts=dict(services=all_services))
if incomplete_warning:
results['msg'] = "WARNING: Could not find status for all services. Sometimes this is due to insufficient privileges."
results[
"msg"
] = "WARNING: Could not find status for all services. Sometimes this is due to insufficient privileges."
module.exit_json(**results)


main()
main()
Loading

0 comments on commit ec8e04c

Please sign in to comment.