Helper functions to create auditd-rulesets

auditd_ruleset.py

import subprocess
from enum import Enum


class AuditAction(str, Enum):
    always = "always"
    never = "never"

    def __str__(self):
        return str.__str__(self)


class AuditList(str, Enum):
    exit_ = "exit"
    user = "user"

    def __str__(self):
        return str.__str__(self)


class AuditPermissionException(Exception):
    pass


class AuditPermissions:

    r = "r"
    w = "w"
    x = "x"
    a = "a"

    def __init__(self, permissions="rwxa"):
        if len(permissions) < 1 or len(permissions) > 4:
            raise AuditPermissionException("Invalid permission specified")
        
        self.perm = []
        
        for p in permissions:
            if p == self.r:
                self.perm.append(self.r)
            elif p == self.w:
                self.perm.append(self.w)
            elif p == self.x:
                self.perm.append(self.x)
            elif p == self.a:
                self.perm.append(self.a)
            else:
                raise AuditPermissionException("Invalid permission specified")

    def __str__(self):
        return "".join(self.perm)


class AuditRule:

    @staticmethod
    def construct_file_watch(f, perm, fields=None, keyname=None):               
        if not isinstance(perm, AuditPermissions):
            perm = AuditPermissions(perm)
            
        base_rule = f"-w {f} -p {perm}"

        if fields:
            for f in fields:
                base_rule += f" -F {f}"

        if keyname:
            base_rule += f" -k {keyname}"

        return base_rule

    @staticmethod
    def construct_syscall_rule(act, ls, sc, fields=None, keyname=None):
        if isinstance(sc, int):
            sc = str(sc)

        base_rule = f"-a {act},{ls} -S {sc}"

        if fields:
            for f in fields:
                base_rule += f" -F {f}"
        if keyname:
            base_rule += f" -k {keyname}"

        return base_rule


# Utility function
def get_syscall_tuples(sc_hdr_file="/usr/include/x86_64-linux-gnu/asm/unistd_64.h"):
    proc = subprocess.Popen(["cat", sc_hdr_file], stdout=subprocess.PIPE)
    out, err = proc.communicate()
    syscalls = [s for s in out.decode("utf-8").splitlines() if s]
    define = "#define __NR_"
    syscalls = [s.replace(define, "").split(" ") for s in syscalls[2:-1]]

    return syscalls