File permissions decide who can read and execute a script. Sometimes, a finer control is needed: A script installed in a shared directory should only be run by a team, or only by a service account. In this step, disk_watch.py is restricted to the members of the ops group.
#!/usr/bin/env python3
"""
--------------------------------------------------------------------------------
@author : Luiggi Vercotti
@date : 2026-10-08
@version : 0.9.0
@description : Monitor filesystem usage
--------------------------------------------------------------------------------
Update:
0.1.0 2026-10-08 - Luiggi Vercotti - cre: Script creation
0.2.0 2026-10-08 - Luiggi Vercotti - evo: Several mount points, log levels
0.3.0 2026-10-08 - Luiggi Vercotti - evo: Monitoring exit status
0.4.0 2026-10-08 - Luiggi Vercotti - evo: Configuration file
0.5.0 2026-10-08 - Luiggi Vercotti - evo: Command line options
0.6.0 2026-10-08 - Luiggi Vercotti - evo: Alerting API token
0.7.0 2026-10-08 - Luiggi Vercotti - evo: Report file
0.8.0 2026-10-08 - Luiggi Vercotti - evo: Error handling
0.9.0 2026-10-08 - Luiggi Vercotti - evo: Access control
--------------------------------------------------------------------------------
List of authorized users or group:
@group:ops
--------------------------------------------------------------------------------
List of mandatory configuration parameters:
@conf:filesystems|mount_points|str|false
@conf:thresholds|warning|int|false
@conf:thresholds|critical|int|false
@conf:alerting|token|str|true
--------------------------------------------------------------------------------
List of execution options and argument parameters:
@args:mount-point|str|Mount points to check (default: from configuration)|+||exclude||false|false
@args:exclude|str|Mount points to skip|+||||false|false
@args:warning|int|Warning threshold in percent (default: from configuration)|1||||false|false
@args:critical|int|Critical threshold in percent (default: from configuration)|1||||false|false
@args:token|str|Alerting API token (default: from configuration)|1||||false|true
@args:report|bool|Write a report into the report directory||false|||false|false
@args:format|choice|Report format (default: text)|1|text,csv||report|false|false
--------------------------------------------------------------------------------
Functioning:
---------------
Checks the usage of each mount point against a warning and a critical
threshold and exits with a monitoring compatible exit status:
0: OK, 1: WARNING, 2: CRITICAL, 3: UNKNOWN (a mount point can not be checked)
Configuration:
- filesystems::mount_points: Comma separated list of mount points
- thresholds::warning: Warning threshold in percent
- thresholds::critical: Critical threshold in percent
- alerting::token: Alerting API token (secret)
"""
# ------------------------------------------------------------------------------
# Initialization of the environment
# ------------------------------------------------------------------------------
import os
import sys
import shutil
import datetime
import scrippy_core
from scrippy_core import logger, ScrippyCoreError
OK, WARNING, CRITICAL, UNKNOWN = 0, 1, 2, 3
STATUS = {OK: "OK", WARNING: "WARNING", CRITICAL: "CRITICAL", UNKNOWN: "UNKNOWN"}
EXTENSIONS = {"text": "txt", "csv": "csv"}
# ------------------------------------------------------------------------------
# Definition of functions and classes
# ------------------------------------------------------------------------------
def get_mount_points(config, args):
if args.mount_point:
return args.mount_point
mount_points = [mp.strip() for mp in config.get("filesystems", "mount_points").split(",")]
return [mp for mp in mount_points if mp not in (args.exclude or [])]
def get_thresholds(config, args):
warning = args.warning if args.warning is not None else config.get("thresholds", "warning", "int")
critical = args.critical if args.critical is not None else config.get("thresholds", "critical", "int")
if not 0 < warning < critical <= 100:
raise ScrippyCoreError(f"Invalid thresholds: warning={warning}, critical={critical}")
return warning, critical
def get_usage(mount_point):
logger.debug(f"[+] Checking {mount_point}")
usage = shutil.disk_usage(mount_point)
return round(usage.used * 100 / usage.total, 1)
def check(mount_points, warning, critical):
results = []
for mount_point in mount_points:
try:
usage = get_usage(mount_point)
except OSError as err:
logger.error(f" '-> {mount_point}: {err.strerror}")
results.append((mount_point, None, UNKNOWN))
continue
if usage >= critical:
logger.error(f" '-> {mount_point}: {usage}% [CRITICAL]")
status = CRITICAL
elif usage >= warning:
logger.warning(f" '-> {mount_point}: {usage}% [WARNING]")
status = WARNING
else:
logger.info(f" '-> {mount_point}: {usage}% [OK]")
status = OK
results.append((mount_point, usage, status))
return results
def write_report(results, report_format, workspace_path, reportdir):
name = f"disk_watch_{datetime.datetime.now():%Y%m%d_%H%M%S}.{EXTENSIONS[report_format]}"
tmp_report = os.path.join(workspace_path, name)
logger.info(f"[+] Writing report: {tmp_report}")
with open(tmp_report, mode="w", encoding="utf-8") as report:
for mount_point, usage, status in results:
if report_format == "csv":
report.write(f"{mount_point};{usage};{STATUS[status]}\n")
else:
report.write(f"{mount_point:<20} {usage}% {STATUS[status]}\n")
report_dir = os.path.join(reportdir, "disk_watch")
os.makedirs(report_dir, exist_ok=True)
logger.info(f"[+] Publishing report into {report_dir}")
shutil.move(tmp_report, os.path.join(report_dir, name))
# ------------------------------------------------------------------------------
# Main processing
# ------------------------------------------------------------------------------
def main():
with scrippy_core.ScriptContext(retention=100) as _context:
config = _context.config
args = _context.args
token = args.token or config.get("alerting", "token")
logger.debug(f"[+] Alerting API token: {token}")
warning, critical = get_thresholds(config, args)
logger.log_section("disk_watch", f"Thresholds: warning={warning}% critical={critical}%")
results = check(get_mount_points(config, args), warning, critical)
if args.report:
write_report(results, args.format or "text", _context.workspace_path, _context.env["reportdir"])
status = max(status for _, _, status in results)
logger.info(f"[+] Status: {STATUS[status]}")
if status != OK:
sys.exit(status)
# ------------------------------------------------------------------------------
# Entry point
# ------------------------------------------------------------------------------
if __name__ == '__main__':
main()
What changed: The List of authorized users or group section declares @group:ops.
When run by a user who is not a member of the ops group, the script stops before running any of your code:
[...]
[20261008 19:15:12] [CRITICAL] [ScrippyCoreError]: [BadGroupError] User is not member of an authorized group
Note: If the ops group does not exist on your system, the script stops with [UnknownGroupError] Unknown group: "getgrnam(): name not found: 'ops'". Create the group, or replace ops by one of your groups (see the id command) to run this step.
@user:<username> restricts the execution to the listed users. Repeat the declaration to allow several users:
@user:luiggi
@user:harry
@group:<group name> restricts the execution to the members of the listed groups. Repeat the declaration to allow several groups:
@group:ops
@group:dba
When both @user and @group are declared, both conditions must be met: The user must be one of the listed users and a member of one of the listed groups.
@user:luiggi
@user:harry
@group:ops
With the declarations above, luiggi can run the script only while being a member of the ops group, and the other members of ops can not run it at all.
| Error | Cause |
|---|---|
[BadUserError] Unauthorized user |
The user is not one of the @user users |
[BadGroupError] User is not member of an authorized group |
The user is not a member of any of the @group groups |
[UnknownUserError] Unknown user: ... |
A @user user does not exist on the system |
[UnknownGroupError] Unknown group: ... |
A @group group does not exist on the system |
These checks are done when entering the execution context. A refused execution is not recorded in the history.
@user nor @group declarations, any user allowed by the file permissions can run the script.sudo -u svc_monitor, the user checked is svc_monitor. The history records both the original user (Origin) and the effective one (Exec).Warning: The header is part of the script: Anyone who can modify the script can modify the declarations. Make sure the script file is only writable by its maintainers.
What happens if disk_watch.py is scheduled every minute and one execution takes longer than expected? In the last step, you will prevent simultaneous executions.