Mirror
/
LSMS
mirror of https://github.com/sqall01/LSMS.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

sync internal repo

This commit is contained in:
Andre Pawlowski 2022-02-10 16:12:03 +01:00
parent ed87b33550
commit 8f8497f63e
8 changed files with 352 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@ -34,6 +34,7 @@ If you want to use the scripts to monitor your Linux system constantly, you have
| Name | Script |
|---------------------------------------------|--------------------------------------------------------------------------|
| Monitoring cron files | [monitor_cron.py](scripts/monitor_cron.py) |
| Monitoring /etc/hosts file | [monitor_hosts_file.py](scripts/monitor_hosts_file.py) |
| Monitoring /etc/ld.so.preload file | [monitor_ld_preload.py](scripts/monitor_ld_preload.py) |
| Monitoring /etc/passwd file | [monitor_passwd.py](scripts/monitor_passwd.py) |

10
scripts/config/monitor_cron.py Normal file
View File

@ -0,0 +1,10 @@
from typing import List
# Is the script allowed to run or not?
ACTIVATED = True
# Directory in which cron stores crontab files for individual users. Following is the default on Ubuntu/Debian.
USER_CRONTAB_DIR = "/var/spool/cron/crontabs/"
# Directories in which scripts can be placed that are executed by cron. Following list are the defaults on Ubuntu/Debian.
CRON_SCRIPT_DIRS = ["/etc/cron.daily", "/etc/cron.hourly", "/etc/cron.monthly", "/etc/cron.weekly"] # type: List[str]

6
scripts/lib/state.py
View File

@ -5,11 +5,7 @@ from typing import Dict, Any
class StateException(Exception):
def __init__(self, msg: str):
self._msg = msg
def __str__(self):
return self._msg
pass
def load_state(state_dir: str) -> Dict[str, Any]:

91
scripts/lib/util_user.py Normal file
View File

@ -0,0 +1,91 @@
from typing import List
class PasswdException(Exception):
pass
class SystemUser:
def __init__(self,
name: str,
password: str,
uid: int,
gid: int,
info: str,
home: str,
shell: str):
self._name = name
self._password = password
self._uid = uid
self._gid = gid
self._info = info
self._home = home
self._shell = shell
def __str__(self):
return "%s:%s:%d:%d:%s:%s:%s" % (self._name,
self._password,
self._uid,
self._gid,
self._info,
self._home,
self._shell)
@staticmethod
def from_passwd_line(passwd_line: str):
line_split = passwd_line.split(":")
return SystemUser(line_split[0],
line_split[1],
int(line_split[2]),
int(line_split[3]),
line_split[4],
line_split[5],
line_split[6])
@property
def name(self) -> str:
return self._name
@property
def password(self) -> str:
return self._password
@property
def uid(self) -> int:
return self._uid
@property
def gid(self) -> int:
return self._gid
@property
def info(self) -> str:
return self._info
@property
def home(self) -> str:
return self._home
@property
def shell(self) -> str:
return self._shell
def get_system_users() -> List[SystemUser]:
"""
Gets the system's users from /etc/passwd
:return:
"""
user_list = []
try:
with open("/etc/passwd", 'rt') as fp:
for line in fp:
if line.strip() == "":
continue
user_list.append(SystemUser.from_passwd_line(line.strip()))
except Exception as e:
raise PasswdException(str(e))
return user_list

239
scripts/monitor_cron.py Executable file
View File

@ -0,0 +1,239 @@
#!/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 /etc/crontab, /etc/cron.d/*, user specific crontab files and script files run by cron (e.g., script files in /etc/cron.hourly) for changes to detect attempts for attacker persistence.
Additionally, check if crontab entries and user specific crontab files belong to existing system users.
NOTE: The first execution of this script will only show you the current state of the environment which should be acknowledged before monitoring for changes will become an effective security measure.
Requirements:
None
"""
import hashlib
import os
import re
from typing import Dict, List, Set
from lib.state import load_state, store_state
from lib.util import output_error, output_finding
from lib.util_user import get_system_users
# Read configuration.
try:
from config.config import ALERTR_FIFO, FROM_ADDR, TO_ADDR, STATE_DIR
from config.monitor_cron import ACTIVATED, USER_CRONTAB_DIR, CRON_SCRIPT_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__))
USER_CRONTAB_DIR = "/var/spool/cron/crontabs/"
CRON_SCRIPT_DIRS = ["/etc/cron.daily", "/etc/cron.hourly", "/etc/cron.monthly", "/etc/cron.weekly", "/etc/cron.d"]
def _calculate_hash(file_location: str) -> str:
with open(file_location, "rb") as fp:
file_hash = hashlib.md5()
chunk = fp.read(1048576)
while chunk:
file_hash.update(chunk)
chunk = fp.read(1048576)
return file_hash.hexdigest().upper()
def _get_cron_script_files() -> Dict[str, str]:
cron_script_files = dict()
for cron_script_dir in CRON_SCRIPT_DIRS:
for cron_script_file in os.listdir(cron_script_dir):
cron_script_location = os.path.join(cron_script_dir, cron_script_file)
cron_script_files[cron_script_location] = _calculate_hash(cron_script_location)
return cron_script_files
def _get_crontab_files() -> Dict[str, List[str]]:
crontab_entries = dict()
# Add default location of crontab entries.
crontab_files = ["/etc/crontab"]
# Add crontab files that are installed by other software.
for crond_file in os.listdir("/etc/cron.d"):
crond_location = os.path.join("/etc/cron.d", crond_file)
if os.path.isfile(crond_location):
crontab_files.append(crond_location)
# Add user individual crontab files.
for crontab_file in os.listdir(USER_CRONTAB_DIR):
crontab_location = os.path.join(USER_CRONTAB_DIR, crontab_file)
if os.path.isfile(crontab_location):
crontab_files.append(crontab_location)
for crontab_file in crontab_files:
crontab_entries[crontab_file] = _parse_crontab(crontab_file)
return crontab_entries
def _get_crontab_users(curr_crontab_data: Dict[str, List[str]]) -> Set[str]:
crontab_users = set()
# User individual crontab files are named after the username.
for crontab_file in os.listdir(USER_CRONTAB_DIR):
crontab_users.add(crontab_file)
# Extract all crontab entries that contain a user that should run the command.
crontab_entries_with_user = list(curr_crontab_data["/etc/crontab"])
for crontab_file in curr_crontab_data.keys():
if crontab_file.startswith("/etc/cron.d/"):
crontab_entries_with_user.extend(curr_crontab_data[crontab_file])
# Extract user from crontab entries.
for crontab_entry in crontab_entries_with_user:
match = re.fullmatch(r'([*?\-,/\d]+)\s([*?\-,/\d]+)\s([*?\-,/\dLW]+)\s([*?\-,/\d]+)\s([*?\-,/\dL#]+)\s([\-\w]+)\s+(.+)',
crontab_entry)
if match is not None:
crontab_users.add(match.groups()[5])
return crontab_users
def _parse_crontab(file_location: str) -> List[str]:
entries = list()
with open(file_location, 'rt') as fp:
for line in fp:
line_strip = line.strip()
if line_strip == "" or line_strip[0] == "#":
continue
entries.append(line_strip)
return entries
def monitor_cron():
# 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_cron_data = {}
try:
stored_cron_data = load_state(STATE_DIR)
except Exception as e:
output_error(__file__, str(e))
return
# Add crontab key in case we do not have any stored data yet.
if "crontab" not in stored_cron_data.keys():
stored_cron_data["crontab"] = {}
# Add cronscripts key in case we do not have any stored data yet.
if "cronscripts" not in stored_cron_data.keys():
stored_cron_data["cronscripts"] = {}
curr_crontab_data = {}
try:
curr_crontab_data = _get_crontab_files()
except Exception as e:
output_error(__file__, str(e))
return
# Compare stored crontab data with current one.
stored_crontab_data = stored_cron_data["crontab"]
for stored_crontab_file, stored_crontab_entries in stored_crontab_data.items():
# Check if crontab file was deleted.
if stored_crontab_file not in curr_crontab_data.keys():
message = "Crontab file '%s' was deleted." % stored_crontab_file
output_finding(__file__, message)
continue
# Check entries were deleted.
for stored_crontab_entry in stored_crontab_entries:
if stored_crontab_entry not in curr_crontab_data[stored_crontab_file]:
message = "Entry in crontab file '%s' was deleted.\n\n" % stored_crontab_file
message += "Deleted entry: %s" % stored_crontab_entry
output_finding(__file__, message)
# Check entries were added.
for curr_crontab_entry in curr_crontab_data[stored_crontab_file]:
if curr_crontab_entry not in stored_crontab_entries:
message = "Entry in crontab file '%s' was added.\n\n" % stored_crontab_file
message += "Added entry: %s" % curr_crontab_entry
output_finding(__file__, message)
# Check new crontab file added.
for curr_crontab_file, curr_crontab_entries in curr_crontab_data.items():
if curr_crontab_file not in stored_crontab_data.keys():
message = "Crontab file '%s' was added.\n\n" % curr_crontab_file
for curr_crontab_entry in curr_crontab_entries:
message += "Entry: %s\n" % curr_crontab_entry
output_finding(__file__, message)
# Check users running crontab entries actually exist as system users.
system_users = get_system_users()
for crontab_user in _get_crontab_users(curr_crontab_data):
if not any([crontab_user == x.name for x in system_users]):
message = "Crontab entry or entries are run as user '%s' but no such system user exists." % crontab_user
output_finding(__file__, message)
curr_script_data = {}
try:
curr_script_data = _get_cron_script_files()
except Exception as e:
output_error(__file__, str(e))
return
# Compare stored cron script data with current one.
stored_script_data = stored_cron_data["cronscripts"]
for stored_script_file, stored_script_hash in stored_script_data.items():
# Check if cron script file was deleted.
if stored_script_file not in curr_script_data.keys():
message = "Cron script file '%s' was deleted." % stored_script_file
output_finding(__file__, message)
continue
# Check if cron script file was modified.
if stored_script_hash != curr_script_data[stored_script_file]:
message = "Cron script file '%s' was modified." % stored_script_file
output_finding(__file__, message)
# Check new cron script file added.
for curr_script_file in curr_script_data.keys():
if curr_script_file not in stored_script_data.keys():
message = "Cron script file '%s' was added." % curr_script_file
output_finding(__file__, message)
try:
store_state(STATE_DIR, {"crontab": curr_crontab_data,
"cronscripts": curr_script_data})
except Exception as e:
output_error(__file__, str(e))
if __name__ == '__main__':
monitor_cron()

6
scripts/monitor_hosts_file.py
View File

@ -37,11 +37,7 @@ except:
class MonitorHostsException(Exception):
def __init__(self, msg: str):
self._msg = msg
def __str__(self):
return self._msg
pass
def _get_hosts() -> Dict[str, Set[str]]:

16
scripts/monitor_passwd.py
View File

@ -22,6 +22,7 @@ from typing import Dict
from lib.state import load_state, store_state
from lib.util import output_error, output_finding
from lib.util_user import get_system_users
# Read configuration.
try:
@ -37,19 +38,10 @@ except:
def _get_passwd() -> Dict[str, str]:
passwd_data = {}
with open("/etc/passwd", 'rt') as fp:
for line in fp:
line = line.strip()
if line == "":
continue
entry = line.strip().split(":")
user = entry[0]
passwd_data[user] = line
for user_obj in get_system_users():
user = user_obj.name
passwd_data[user] = str(user_obj)
return passwd_data

29
scripts/monitor_ssh_authorized_keys.py
View File

@ -23,6 +23,7 @@ from typing import List, Tuple, Dict, Any
from lib.state import load_state, store_state
from lib.util import output_error, output_finding
from lib.util_user import get_system_users
# Read configuration.
try:
@ -38,30 +39,16 @@ except:
class MonitorSSHException(Exception):
def __init__(self, msg: str):
self._msg = msg
def __str__(self):
return self._msg
pass
def _get_home_dirs_from_passwd() -> List[Tuple[str, str]]:
user_home_list = []
try:
with open("/etc/passwd", 'rt') as fp:
for line in fp:
line_split = line.split(":")
user_home_list.append((line_split[0], line_split[5]))
except Exception as e:
raise MonitorSSHException(str(e))
return user_home_list
def _get_home_dirs() -> List[Tuple[str, str]]:
return [(x.name, x.home) for x in get_system_users()]
def _get_system_ssh_data() -> List[Dict[str, Any]]:
ssh_data = []
user_home_list = _get_home_dirs_from_passwd()
user_home_list = _get_home_dirs()
for user, home in user_home_list:
# Monitor "authorized_keys2" too since SSH also checks this file for keys (even though it is deprecated).
@ -186,9 +173,3 @@ def monitor_ssh_authorized_keys():
if __name__ == '__main__':
monitor_ssh_authorized_keys()