From 4011007b445e8f8da9b0cc45eccd793b94f6b5ce Mon Sep 17 00:00:00 2001
From: Sergio Correia <scorreia@redhat.com>
Date: Thu, 29 Jul 2021 19:25:43 -0300
Subject: [PATCH] Add ausysrulevalidate
---
contrib/ausysrulevalidate | 198 ++++++++++++++++++++++++++++++++++++++
1 file changed, 198 insertions(+)
create mode 100755 contrib/ausysrulevalidate
diff --git a/contrib/ausysrulevalidate b/contrib/ausysrulevalidate
new file mode 100755
index 0000000..a251b2c
--- /dev/null
+++ b/contrib/ausysrulevalidate
@@ -0,0 +1,198 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+# ausysrulevalidate - A program that lets you validate the syscalls
+# in audit rules.
+# Copyright (c) 2021 Red Hat Inc., Durham, North Carolina.
+# All Rights Reserved.
+#
+# This software may be freely redistributed and/or modified under the
+# terms of the GNU General Public License as published by the Free
+# Software Foundation; either version 2, or (at your option) any
+# later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; see the file COPYING. If not, write to the
+# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor
+# Boston, MA 02110-1335, USA.
+#
+# Authors:
+# Sergio Correia <scorreia@redhat.com>
+
+""" This program lets you validate syscalls in audit rules. """
+
+import argparse
+import os.path
+import sys
+
+import audit
+
+
+class AuSyscallRuleValidate:
+ """AuSyscallRuleValidate validates syscalls in audit rules."""
+
+ def __init__(self):
+ self.syscalls_table = {}
+ self.invalid_syscalls = {}
+ self.machines = {
+ "b32": audit.audit_determine_machine("b32"),
+ "b64": audit.audit_determine_machine("b64"),
+ }
+
+ if self.machines["b32"] == -1 or self.machines["b64"] == -1:
+ sys.stderr.write("ERROR: Unable to determine machine type\n")
+ sys.exit(1)
+
+ def validate_syscall(self, arch, syscall):
+ """Validates a single syscall."""
+
+ if syscall == "all":
+ return True
+
+ lookup = "{0}:{1}".format(arch, syscall)
+ if lookup in self.syscalls_table:
+ return self.syscalls_table[lookup]
+
+ ret = audit.audit_name_to_syscall(syscall, self.machines[arch])
+ self.syscalls_table[lookup] = ret != -1
+ if not self.syscalls_table[lookup]:
+ self.invalid_syscalls[lookup] = lookup
+
+ return self.syscalls_table[lookup]
+
+ def process_syscalls(self, arch, syscalls):
+ """Processes a group of syscalls, validating them individually."""
+
+ scalls = syscalls.split(",")
+ processed = []
+ for syscall in scalls:
+ if self.validate_syscall(arch, syscall):
+ processed.append(syscall)
+ return ",".join(processed)
+
+ def parse_line(self, line):
+ """Processes a single line from the audit rules file, and returns the
+ same line adjusted, if required, by removing invalid syscalls, or even
+ removing the rule altogether, if no valid syscall remain after
+ validation."""
+
+ if line.lstrip().startswith("#") or "-S" not in line:
+ return line
+
+ # We do have a rule specifying syscalls, so let's validate them.
+ tokens = line.split()
+ processed = []
+ is_syscall = False
+ arch = None
+
+ for val in tokens:
+ if not is_syscall:
+ processed.append(val)
+
+ if val.startswith("arch="):
+ archs = val.split("=")
+ if len(archs) == 2:
+ arch = val.split("=")[1]
+ if arch not in self.machines:
+ sys.stderr.write("ERROR: unexpected arch '{0}'\n".format(arch))
+ continue
+
+ if val == "-S":
+ is_syscall = True
+ continue
+
+ if is_syscall:
+ is_syscall = False
+ scalls = self.process_syscalls(arch, val)
+
+ if len(scalls) == 0:
+ processed = processed[:-1]
+ continue
+ processed.append(scalls)
+
+ if "-S" not in processed:
+ # Removing rule altogether, as we have no valid syscalls remaining.
+ return None
+ return " ".join(processed)
+
+ def process_rules(self, rules_file):
+ """Reads a file with audit rules and returns the rules after
+ validation of syscalls/architecture. Invalid syscalls will be removed
+ and, if there are no valid remaining syscalls, the rule itself is
+ removed."""
+
+ if not os.path.isfile(rules_file):
+ sys.stderr.write("ERROR: rules file '{0}' not found\n".format(rules_file))
+ sys.exit(1)
+
+ with open(rules_file) as rules:
+ content = rules.readlines()
+
+ processed = []
+ changed = False
+ for line in content:
+ validated = self.parse_line(line)
+ if validated is None:
+ changed = True
+ continue
+
+ if validated.rstrip("\r\n") != line.rstrip("\r\n"):
+ changed = True
+ processed.append(validated.rstrip("\r\n"))
+
+ invalid_syscalls = []
+ for invalid in self.invalid_syscalls:
+ invalid_syscalls.append(invalid)
+
+ return (processed, changed, invalid_syscalls)
+
+ def update_rules(self, rules_file):
+ """Reads a file with audit rules and updates it after validation of
+ syscalls/architecture. Invalid syscalls will be removed and, if
+ there are no valid remaining syscalls, the rule itself is removed."""
+
+ new_rules, changed, invalid_syscalls = self.process_rules(rules_file)
+ if changed:
+ with open(rules_file, "w") as rules:
+ for line in new_rules:
+ rules.write("{0}\n".format(line))
+
+ return (new_rules, changed, invalid_syscalls)
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser(description="ausysrulevalidate")
+ parser.add_argument(
+ "-u", "--update", help="Update rules file if required", action="store_true"
+ )
+ parser.add_argument(
+ "-v", "--verbose", help="Show the resulting rules file", action="store_true"
+ )
+ required_named = parser.add_argument_group("required named arguments")
+ required_named.add_argument(
+ "-r", "--rules-file", help="Rules file name", required=True
+ )
+ args = parser.parse_args()
+
+ validator = AuSyscallRuleValidate()
+
+ action = validator.process_rules
+ if args.update:
+ action = validator.update_rules
+
+ data, changed, invalid = action(args.rules_file)
+ if changed:
+ verb = "require"
+ if args.update:
+ verb += "d"
+ sys.stderr.write("Rules in '{0}' {1} changes\n".format(args.rules_file, verb))
+ if len(invalid) > 0:
+ sys.stderr.write("Invalid syscalls: {0}\n".format(", ".join(invalid)))
+
+ if args.verbose:
+ print(*data, sep="\n")
--
2.31.1