GUI editor to tame mod_security rules

⌈⌋ ⎇ branch:  modseccfg


Check-in [21714627ef]

Many hyperlinks are disabled.
Use anonymous login to enable hyperlinks.

Overview
Comment:Fix is_glob taking precedence over not writable() in log reader, more details for subprocess.run() tasks (instead of os.system() now)
Downloads: Tarball | ZIP archive | SQL archive
Timelines: family | ancestors | descendants | both | trunk
Files: files | file ages | folders
SHA3-256: 21714627ef8d2d8eb50e0d4a4ba087a2ce813229db8fd43acf8764f1b6331bcc
User & Date: mario 2020-11-26 00:27:07
Context
2020-11-26
15:29
Turn conf in DictObj() check-in: 87673cca9b user: mario tags: trunk
00:27
Fix is_glob taking precedence over not writable() in log reader, more details for subprocess.run() tasks (instead of os.system() now) check-in: 21714627ef user: mario tags: trunk
00:26
Release as 0.4.0 check-in: 393a1fb162 user: mario tags: trunk, 0.4.0
Changes
Hide Diffs Unified Diffs Ignore Whitespace Patch

Changes to modseccfg/logs.py.

︙︙
85
86
87
88
89
90
91

92
93
94
95
96
97
98
99
100
101
102





103
104
105
106
107
108
109

    )
rx.audit_all = re.compile("|".join(rx.audit), re.I|re.M)


# search through log file, filter, extract rule ids, return list of log lines
def scan_log(fn="", pipe=None):


    if fn == state.log_curr:
        return   # no update // notably this will prevent File > Rescan Logs
    state.log_curr = ""
    if not srvroot.exists(fn):
        return
    log_curr = fn
    log_count = {}
    
    # type
    is_glob = re.search("\*(/?\*)+", fn)
    is_audit = is_glob or re.search("audit", fn)





    
    # handle
    if pipe:
        pass
    elif is_glob:
        pipe = open_glob(fn)
    elif fn:








>



<
<
<
<
<



>
>
>
>
>








85
86
87
88
89
90
91
92
93
94
95





96
97
98
99
100
101
102
103
104
105
106
107
108
109
110

    )
rx.audit_all = re.compile("|".join(rx.audit), re.I|re.M)


# search through log file, filter, extract rule ids, return list of log lines
def scan_log(fn="", pipe=None):

    #print(f"scan_log(fn='{fn}')")
    if fn == state.log_curr:
        return   # no update // notably this will prevent File > Rescan Logs
    state.log_curr = ""





    # type
    is_glob = re.search("\*(/?\*)+", fn)
    is_audit = is_glob or re.search("audit", fn)
    if not srvroot.exists(fn) and not is_glob:
        return
    state.log_curr = fn
    log_count = {}
    
    
    # handle
    if pipe:
        pass
    elif is_glob:
        pipe = open_glob(fn)
    elif fn:
︙︙

Changes to modseccfg/utils.py.

︙︙
118
119
120
121
122
123
124




125


126


127
128
129

130
131

132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

152
153


154
155
156
157
158
159
160

            self.srv = self.srvname = ""
        else:
            self.local = 0
            self.srvname = re.sub(":.*?$", "", srv[0])
            self.srv = self.srvname + ":/"   # must be / root
            self.mnt = expandpath(conf["sshfs_mount"]) + "/" + self.srv
            os.makedirs(self.mnt, 0o0700, True)




            sshfs_o = re.sub("(^\s*(?=\w)|-o\s*|[^\w=]+)", "-o ", conf.get("sshfs_o", ""))


            os.system(f"sshfs {sshfs_o} {self.srv}/ {self.mnt}")


            atexit.register(self.umount)

    def umount(self):

        if self.mnt and self.srv:
            os.system("fusermount -u " + self.mnt)

            os.rmdir(self.mnt)

    def fn(self, fn):
        return self.mnt + fn

    def read(self, fn):
        if not self.exists(fn):
            if not re.search("letsencrypt|ssl", fn):
                print("WARNING: file not found", self.mnt, fn)
            return ""
        with open(self.fn(fn), "r", encoding="utf8") as f:
            return f.read()

    def write(self, fn, src):
        with open(self.fn(fn), "w", encoding="utf8") as f:
            return f.write(src)

    def popen(self, cmd, action="r"):
        if not self.local:
            cmd = ["ssh", self.srvname] + cmd

        if action=="r":
            return subprocess.Popen(cmd, stdout=subprocess.PIPE).stdout


        else:
            return subprocess.Popen(cmd, stdin=subprocess.PIPE).stdin

    def exists(self, fn):
        return os.path.exists(self.fn(fn))

    def writable(self, fn):








>
>
>
>
|
>
>
|
>
>



>

|
>




















>


>
>








118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173

            self.srv = self.srvname = ""
        else:
            self.local = 0
            self.srvname = re.sub(":.*?$", "", srv[0])
            self.srv = self.srvname + ":/"   # must be / root
            self.mnt = expandpath(conf["sshfs_mount"]) + "/" + self.srv
            os.makedirs(self.mnt, 0o0700, True)
            self.mount()

    def mount(self):
        opts = []
        for opt in re.findall("(?<!-)\\b[\w=]+", conf.get("sshfs_o", "")):
            opts += ["-o", opt]
        cmd = ["sshfs"] + opts + [self.srv+"/", self.mnt]
        if self.mnt and self.srv:
            print(f"srvroot.mount = {cmd}")
            subprocess.run(cmd)
            atexit.register(self.umount)

    def umount(self):
        cmd = ["fusermount", "-u", self.mnt]
        if self.mnt and self.srv:
            print(f"srvroot.umount = {cmd}")
            subprocess.run(cmd)
            os.rmdir(self.mnt)

    def fn(self, fn):
        return self.mnt + fn

    def read(self, fn):
        if not self.exists(fn):
            if not re.search("letsencrypt|ssl", fn):
                print("WARNING: file not found", self.mnt, fn)
            return ""
        with open(self.fn(fn), "r", encoding="utf8") as f:
            return f.read()

    def write(self, fn, src):
        with open(self.fn(fn), "w", encoding="utf8") as f:
            return f.write(src)

    def popen(self, cmd, action="r"):
        if not self.local:
            cmd = ["ssh", self.srvname] + cmd
            print(f"srvroot.popen = {cmd}")
        if action=="r":
            return subprocess.Popen(cmd, stdout=subprocess.PIPE).stdout
        elif action=="e":
            return subprocess.Popen(cmd, stderr=subprocess.PIPE).stderr
        else:
            return subprocess.Popen(cmd, stdin=subprocess.PIPE).stdin

    def exists(self, fn):
        return os.path.exists(self.fn(fn))

    def writable(self, fn):
︙︙