-
-
Notifications
You must be signed in to change notification settings - Fork 41
/
Copy pathmonitor_systemd_units.py
executable file
·173 lines (140 loc) · 7.14 KB
/
monitor_systemd_units.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
#!/usr/bin/env python3
# written by sqall
# twitter: https://twitter.com/sqall01
# blog: https://h4des.org
# github: https://github.com/sqall01
#
# Licensed under the MIT License.
"""
Short summary:
Monitor systemd unit files to find ones that are used for malware persistence.
NOTE: The first execution of this script should be done with the argument "--init".
Otherwise, the script will only show you the current state of the environment since no state was established yet.
However, this assumes that the system is uncompromised during the initial execution.
Hence, if you are unsure this is the case you should verify the current state
before monitoring for changes will become an effective security measure.
Requirements:
None
Reference:
https://www.trendmicro.com/en_us/research/23/c/iron-tiger-sysupdate-adds-linux-targeting.html
"""
import os
import sys
from typing import Dict
import lib.global_vars
from lib.state import load_state, store_state
from lib.util import get_diff_per_line, output_error, output_finding
# Read configuration.
try:
from config.config import ALERTR_FIFO, FROM_ADDR, TO_ADDR, STATE_DIR
from config.monitor_systemd_units import ACTIVATED, SYSTEMD_UNIT_DIRS
STATE_DIR = os.path.join(os.path.dirname(__file__), STATE_DIR, os.path.basename(__file__))
except:
ALERTR_FIFO = None
FROM_ADDR = None
TO_ADDR = None
ACTIVATED = True
STATE_DIR = os.path.join("/tmp", os.path.basename(__file__))
SYSTEMD_UNIT_DIRS = ["/etc/systemd/system",
"/etc/systemd/user",
"/etc/systemd/network",
"/usr/lib/systemd/system",
"/usr/lib/systemd/user",
"/usr/lib/systemd/network",
"/usr/local/lib/systemd/system",
"/usr/local/lib/systemd/user",
"/usr/local/lib/systemd/network",
"/lib/systemd/system",
"/lib/systemd/user",
"/lib/systemd/network",
"/run/systemd/system"]
def _get_system_unit_files() -> Dict[str, str]:
systemd_unit_files = dict()
for systemd_unit_dir in SYSTEMD_UNIT_DIRS:
for root, _, files in os.walk(systemd_unit_dir):
for file in files:
file_location = os.path.join(root, file)
# Some files are broken symlinks, hence, check if they exist
if os.path.exists(file_location):
with open(file_location, "rt") as fp:
data = fp.read()
# Filter for systemd unit files that can execute commands
if "[Unit]" in data and "[Service]" in data:
# Since keys do not have to start at the beginning of the line, we go through each line,
# remove whitespaces leading whitespaces and check if it starts with a key we are interested in
for line in data.split("\n"):
normalized_line = line.strip()
if any(normalized_line.startswith(x) for x in ["ExecStart",
"ExecStartPre",
"ExecStartPost",
"ExecReload",
"ExecStop",
"ExecStopPost"]):
# Store complete data of unit file. Even on a non-server system such as a
# xubuntu 22.04 we only have around 700 unit files of interest. Calculating with
# 1kB of data per file (which is way larger than a normal unit file has) we only
# need a little over 700 kB memory for this. Even on a Raspberry Pi we have no
# problem doing this. Further, it will prevent race-conditions when we already
# have the data stored and do not read it afterwards from the file if we generate
# alerts.
systemd_unit_files[file_location] = data
break
return systemd_unit_files
def monitor_systemd_units():
# Decide where to output results.
print_output = False
if ALERTR_FIFO is None and FROM_ADDR is None and TO_ADDR is None:
print_output = True
if not ACTIVATED:
if print_output:
print("Module deactivated.")
return
stored_systemd_units_data = {}
try:
stored_systemd_units_data = load_state(STATE_DIR)
except Exception as e:
output_error(__file__, str(e))
return
# Add units key in case we do not have any stored data yet.
if "units" not in stored_systemd_units_data.keys():
stored_systemd_units_data["units"] = {}
curr_systemd_units_data = {}
try:
curr_systemd_units_data = _get_system_unit_files()
except Exception as e:
output_error(__file__, str(e))
return
# Compare stored unit files data with current one.
stored_units_data = stored_systemd_units_data["units"]
for stored_unit_file, stored_unit_data in stored_units_data.items():
# Check if unit file was deleted.
if stored_unit_file not in curr_systemd_units_data.keys():
message = "Systemd unit file '%s' was deleted." % stored_unit_file
output_finding(__file__, message)
continue
# Check if unit file was modified.
if stored_unit_data != curr_systemd_units_data[stored_unit_file]:
diff = get_diff_per_line("Old",
stored_unit_data,
"New",
curr_systemd_units_data[stored_unit_file])
message = "Systemd unit file '%s' was modified:\n\nDiff:\n%s\n\nNew file:\n%s" % (stored_unit_file,
diff,
curr_systemd_units_data[stored_unit_file]) # noqa:E501
output_finding(__file__, message)
# Check new unit file added.
for curr_unit_file in curr_systemd_units_data.keys():
if curr_unit_file not in stored_units_data.keys():
message = "Systemd unit file '%s' was added:\n\n%s" % (curr_unit_file,
curr_systemd_units_data[curr_unit_file])
output_finding(__file__, message)
try:
store_state(STATE_DIR, {"units": curr_systemd_units_data})
except Exception as e:
output_error(__file__, str(e))
if __name__ == '__main__':
if len(sys.argv) == 2:
# Suppress output in our initial execution to establish a state.
if sys.argv[1] == "--init":
lib.global_vars.SUPPRESS_OUTPUT = True
monitor_systemd_units()