From ebae3805f5cb757113f11cca85dfce2f4d14104b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?St=C3=A9phane=20Diemer?= <stephane.diemer@ubicast.eu>
Date: Tue, 13 Nov 2018 08:51:48 +0100
Subject: [PATCH] Use systemctl if dbus is not installed to get fail2ban state.
---
tests/test_fail2ban.py | 55 ++++++++++++++++++++++++++++--------------
1 file changed, 37 insertions(+), 18 deletions(-)
diff --git a/tests/test_fail2ban.py b/tests/test_fail2ban.py
index 14c00ff4..6d30a34e 100755
--- a/tests/test_fail2ban.py
+++ b/tests/test_fail2ban.py
@@ -7,8 +7,6 @@ Checks the current state of the fail2ban service.
from pathlib import Path
import sys
-# pylint: disable=E0401
-import dbus
sys.path.append(str(Path(__file__).parents[1].resolve()))
@@ -25,22 +23,43 @@ def get_service_state(name: str) -> tuple:
:rtype: tuple
"""
- bus = dbus.SystemBus()
- systemd = bus.get_object("org.freedesktop.systemd1", "/org/freedesktop/systemd1")
- manager = dbus.Interface(systemd, "org.freedesktop.systemd1.Manager")
- unit = manager.LoadUnit("{}.service".format(name))
- proxy = bus.get_object("org.freedesktop.systemd1", str(unit))
-
- active = proxy.Get(
- "org.freedesktop.systemd1.Unit",
- "ActiveState",
- dbus_interface="org.freedesktop.DBus.Properties",
- )
- state = proxy.Get(
- "org.freedesktop.systemd1.Unit",
- "SubState",
- dbus_interface="org.freedesktop.DBus.Properties",
- )
+ try:
+ # pylint: disable=E0401
+ import dbus
+ except ImportError:
+ u.log("Using systemctl to get current state.")
+ returncode, output = u.exec_cmd(
+ "systemctl status fail2ban | grep 'Active:'", log_output=False
+ )
+ if returncode != 0:
+ active = "no"
+ state = "not installed"
+ else:
+ info = output.replace("Active:", "").strip().split(" ")
+ if len(info) > 1:
+ active = info[0]
+ state = info[1].strip("()")
+ else:
+ active = "no"
+ state = "unknown"
+ else:
+ u.log("Using dbus to get current state.")
+ bus = dbus.SystemBus()
+ systemd = bus.get_object("org.freedesktop.systemd1", "/org/freedesktop/systemd1")
+ manager = dbus.Interface(systemd, "org.freedesktop.systemd1.Manager")
+ unit = manager.LoadUnit("{}.service".format(name))
+ proxy = bus.get_object("org.freedesktop.systemd1", str(unit))
+
+ active = proxy.Get(
+ "org.freedesktop.systemd1.Unit",
+ "ActiveState",
+ dbus_interface="org.freedesktop.DBus.Properties",
+ )
+ state = proxy.Get(
+ "org.freedesktop.systemd1.Unit",
+ "SubState",
+ dbus_interface="org.freedesktop.DBus.Properties",
+ )
return str(active), str(state)
--
GitLab