| Server IP : 199.250.214.225 / Your IP : 3.145.90.90 Web Server : Apache System : Linux vps64074.inmotionhosting.com 3.10.0-1160.105.1.vz7.214.3 #1 SMP Tue Jan 9 19:45:01 MSK 2024 x86_64 User : nicngo5 ( 1001) PHP Version : 7.4.33 Disable Function : exec,passthru,shell_exec,system MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : OFF Directory : /opt/dedrads/ |
Upload File : |
#!/opt/imh-python/bin/python3
"""Quarantine tool"""
from pathlib import Path
from os import kill, getuid, chown
import re
import pwd
import shlex
import sys
import time
import shutil
import argparse
from argparse import ArgumentTypeError as BadArg
import logging
import logging.handlers
import psutil
import rads
logger = logging.getLogger('quarantine')
USER_RE = re.compile(r'/home[0-9]?/(\w+)')
# EPOCH truncates the minute from the unix timestamp so running this in a loop
# won't create numerous quarantine directories
EPOCH = int(time.time())
EPOCH -= EPOCH % 60
USAGE = """
- PARTIAL: quarantine item1 item2 item3 ...
- FULL: quarantine (--full|-f) user1 user2 ...
- *BOTH: quarantine item1 item2 ... (--full|-f) user1 user2 ...
* Specify items first or else each item will be considered a user"""
DESCRIPTION = """
Relative Paths: If you do not specify the absolute path for a given item, then
it will attempt to create a full path for you. So, the current working
directory is very important. This is because the relative path will be
joined with your current working for directory. For example, running
quarantine public_html from /root will assume that you are trying to
quarantine /root/public_html.
Full Quarantines: The docroots used with full quarantines are found by using the
rads.UserData module. If this fails for some reason, then an error will be
logged. You'll then have to proceed by manually providing the specified
items to be quarantined.
Docroot Recreation: If the actions taken by the quarantine script happen to
quarantine a document root for any domain, then the program will
automatically recreate it.
Specifying Items/Users: There is no limitation on what can be specified as an
item. The script will try to iterate through each item, determine the user,
and quarantine the item as as necessary. If the user or item is invalid, a
warning will be logged and the item will not be quarantined.
Quarantining Subdirectories: When quarantining files, the script creates the
quarantine destination recursively. So, providing the script a subdirectory
item first will cause a File exists error when attempting to quarantine the
top-level directory. Ex: don't do the following:
`quarantine public_html/addondomain.com public_html`
Examples:
quarantine --full userna5
quarantine ~userna5/public_html/addondomain.com ~userna5/subdomain.com
"""
RUN_FIXPERMS = []
def setup_logging(clevel=logging.INFO):
"""Setup logging"""
logger.setLevel(logging.DEBUG)
# stderr
con = logging.StreamHandler(stream=sys.stderr)
con.setLevel(clevel)
con_format = logging.Formatter("%(levelname)s: %(message)s")
con.setFormatter(con_format)
logger.addHandler(con)
# syslog
try:
flog = logging.handlers.SysLogHandler(address='/dev/log')
# syslog is configured to only log INFO and above, so use the same
flog.setLevel(logging.INFO)
flog_format = logging.Formatter("[%(name)s:%(levelname)s] %(message)s")
flog.setFormatter(flog_format)
logger.addHandler(flog)
except Exception as e:
logger.warning("Failed to open logfile: %s", str(e))
def quarantine(path: Path, user_home: Path):
"""Quarantine file/directory"""
# Ex: item = "/home/userna5/public_html"
# Ex: qdest = "/home/userna5/quarantine/quarantine_1538571067/home/userna5/"
# Ex: "/home/userna5/public_html" -> \
# "/home/userna5/quarantine/quarantine_1538571067/home/userna5/public_html"
qdir = user_home / f"quarantine/quarantine_{EPOCH}"
qdest = qdir / str(path.parent).lstrip('/')
try:
qdest.mkdir(mode=0o700, parents=True, exist_ok=True)
except PermissionError:
logger.critical(
"Permission denied to %s: Run fixperms and retry", qdest
)
return
except OSError as exc:
logger.critical("%s: %s", type(exc).__name__, exc)
return
logger.debug("Successfully created quarantine directory (%s).", qdest)
try:
shutil.move(path, qdest)
except Exception as exc:
logger.critical("Error occurred quarantining %s: %s", path, exc)
return
logger.info(
"Successfully quarantined '%s' -> '%s'", path, qdest / path.name
)
def killall(user):
"""Kills all active processes under the passed user"""
procs: dict[int, str] = {}
for pid in psutil.pids():
try:
proc = psutil.Process(pid)
except psutil.NoSuchProcess:
continue
if proc.username() == user:
procs[proc.pid] = shlex.join(proc.cmdline())
killed = []
for pid, cmd in procs.items():
try:
kill(pid, 9)
except OSError as exc:
logging.warning(
"Could not kill %s - %s: %s", cmd, type(exc).__name__, exc
)
continue
killed.append(cmd)
if killed:
print("Killed user processes:", *killed, sep='\n')
def cleanup(user: str, docroots: list[Path]):
"""Miscellaneous cleanup tasks"""
uid = pwd.getpwnam(user).pw_uid
nobody_uid = pwd.getpwnam('nobody').pw_uid
# Iterate through each document root and if it does not
# exist, then re-create it. Since the only root will
# have privileges to set ownership uid:nobody_uid, logic
# exists to skip that step. If the step was skipped, then
# the a WARNING message is shown advising a run of fixperms.
for path in sorted(docroots):
if path.exists():
continue
try:
path.mkdir(parents=True, mode=0o750)
except OSError as exc:
logger.error("Error recreating '%s': %s", path, exc)
continue
logger.debug("Successfully recreated '%s'", path)
# If root, then set proper ownership (user:nobody)
# Else, set flag to notify user they must run fixperms
if UID != 0:
RUN_FIXPERMS.append(user)
continue
try:
chown(path, uid, nobody_uid)
except OSError as e:
logger.error("Error setting ownership for '%s': %s", path, e)
continue
logger.debug("Successfully set ownership for '%s'", path)
def valid_user(user: str):
if not rads.cpuser_safe(user):
raise BadArg(f"{user} is an invalid user")
return user
def valid_path(item: str) -> Path:
path = Path(item)
if not path.resolve().is_relative_to('/home'):
raise BadArg(f"{path} is outside /home")
if not path.exists():
raise BadArg(f"{path} does not exist")
return path
def item_validation(path: Path, user_home: Path):
"""Test for valid item"""
if not path.is_relative_to(user_home):
logger.warning("Not quarantining item (%s) outside homedir", path)
elif not path.exists():
logger.warning("%s wasn't found", path)
elif path == user_home:
logger.warning("Not quarantining user's homedir (%s)", path)
elif not str(path).startswith("/home"):
logger.warning("Quarantining is restricted to /home")
else:
return True
return False
def get_userdata(user: str, all_roots: bool = False) -> list[Path]:
"""Gets docroots from rads.Userdata"""
try:
if all_roots:
docroots = map(Path, rads.UserData(user).all_roots)
else:
docroots = map(Path, rads.UserData(user).merged_roots)
home = Path(f"~{user}").expanduser()
except Exception as exc:
logger.error("Error retrieving %s's cPanel userdata: %s", user, exc)
return []
return [
x
for x in docroots
if x.is_relative_to(home) and str(x).startswith('/home')
]
def parse_args() -> tuple[set[Path], bool]:
"""Parse Arguments"""
parser = argparse.ArgumentParser(
formatter_class=argparse.RawTextHelpFormatter,
usage=USAGE,
description=DESCRIPTION,
)
parser.set_defaults(loglevel=logging.INFO, items=[], users=[])
parser.add_argument(
"items",
nargs='*',
help="files and directories to quarantine",
type=valid_path,
)
parser.add_argument(
"-f",
"--full",
dest='users',
nargs="+",
type=valid_user,
help="Perform a full quarantine of all docroots for given users",
)
parser.add_argument(
"-d",
'--debug',
dest='loglevel',
action='store_const',
const=logging.DEBUG,
help="Enable debug output",
)
parser.add_argument(
"-k",
'--kill',
dest='kill',
action='store_true',
help="Kill all processes under the detected user after quarantine",
)
args = parser.parse_args()
setup_logging(args.loglevel)
items: list[Path] = args.items
for user in args.users: # for any user supplied with --full
items.extend(get_userdata(user))
items = {x.resolve() for x in items}
return items, args.kill
def main():
"""Main Logic"""
paths, do_kill = parse_args()
user_docroots: dict[str, list[Path]] = {}
# For each item: parse user, validate user, store userdata, set
# homedir, validate item, quarantine item. The contents of items can include
# files/directories across multiple users
for path in paths:
if user_match := USER_RE.match(str(path)):
user = user_match.group(1)
else:
logger.warning("Unable to determine user for %s", path)
continue
if not rads.cpuser_safe(user):
logger.warning('Skipping invalid user %s', user)
continue
if user not in user_docroots:
user_docroots[user] = get_userdata(user, all_roots=True)
try:
user_home = Path(f"~{user}").expanduser()
except Exception:
logger.warning("Could not get homedir for %s", user)
continue
if not item_validation(path, user_home):
logger.warning("Skipping %s", path)
continue
quarantine(path, user_home)
if do_kill:
killall(user)
# This block of code calls the cleanup function.
# It currently just ensures that all the user's docroots are present
for user, docroots in user_docroots.items():
cleanup(user, docroots)
if not RUN_FIXPERMS:
return
logger.warning(
"During the quarantine process some document roots were quarantined. "
"You must run fixperms to correct permissions/ownership of these "
"document roots. Please run: fixperms %s",
shlex.join(RUN_FIXPERMS),
)
if __name__ == '__main__':
UID = getuid()
try:
main()
except KeyboardInterrupt:
print("Canceled.")