Lightweight replacement to both epylog and fail2ban.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

159 lines
5.6 KiB

# pyruse is intended as a replacement to both fail2ban and epylog
# Copyright © 2017–2018 Y. Gablin
# Full licensing information in the LICENSE file, or gnu.org/licences/gpl-3.0.txt if the file is missing.
import os
import re
from datetime import datetime
from pyruse.actions.action_dailyReport import Action
from pyruse import config
mail_filename = "email.dump"
wAction = Action({"level": "WARN", "message": "WarnMsg {m}"})
iAction = Action({"level": "INFO", "message": "InfoMsg {m}"})
oAction = Action({"level": "OTHER", "message": "MiscMsg {m}"})
wActFirst = Action({"level": "WARN", "message": "WarnMsg {m}", "details": "FIRST"})
wActLast = Action({"level": "WARN", "message": "WarnMsg {m}", "details": "LAST"})
wActFL = Action({"level": "WARN", "message": "WarnMsg {m}", "details": "FIRSTLAST"})
wActNone = Action({"level": "WARN", "message": "WarnMsg {m}", "details": "NONE"})
def newEntry(m):
return {"__REALTIME_TIMESTAMP": datetime.utcnow(), "m": m}
def whenNewDayThenReport():
if os.path.exists(mail_filename):
os.remove(mail_filename)
oAction.act(newEntry("message1"))
assert not os.path.exists(mail_filename)
Action._hour = 25
oAction.act(newEntry("message2"))
assert os.path.exists(mail_filename)
os.remove(mail_filename)
def whenEmailThenCheck3Sections():
if os.path.exists(mail_filename):
os.remove(mail_filename)
wAction.act(newEntry("messageW"))
iAction.act(newEntry("messageI"))
Action._hour = 25
oAction.act(newEntry("messageO"))
assert os.path.exists(mail_filename)
conf = config.Config().asMap().get("email", {})
reSubject = re.compile(r"^Subject: (.*)")
reFrom = re.compile(r"^From: (.*)")
reTo = re.compile(r"^To: (.*)")
subjOK = False
fromOK = False
toOK = False
nbWarn = 0
nbInfo = 0
nbMisc = 0
with open(mail_filename, 'rt') as m:
for line in m:
match = reSubject.match(line)
if match:
subjOK = match.group(1) == conf.get("subject", "Pyruse Report")
match = reFrom.match(line)
if match:
fromOK = match.group(1) == conf.get("from", "pyruse")
match = reTo.match(line)
if match:
toOK = match.group(1).split(", ") == conf.get("to", ["hostmaster"])
if "WarnMsg" in line:
nbWarn += 1
if "InfoMsg" in line:
nbInfo += 1
if "MiscMsg" in line:
nbMisc += 1
assert subjOK
assert fromOK
assert toOK
assert nbWarn == 2
assert nbInfo == 2
assert nbMisc == 2
os.remove(mail_filename)
def _compareEmailWithExpected(expected):
assert os.path.exists(mail_filename)
reTime = re.compile(r"\d{4}(?:[- :.]\d{2}){6}\d{4}")
warnSeen = False
nbTimes = 0
nbFirst = 0
nbLast = 0
line = ""
with open(mail_filename, 'rt') as m:
for l in m:
if l != "" and l[-1:] == "=":
line += l[:-1]
continue
elif l == "" and warnSeen:
break
line += l
if "WarnMsg" in line:
warnSeen = True
elif not warnSeen:
line = ""
continue
nbTimes += len(reTime.findall(line))
if "From=C2=A0:" in line:
nbFirst += 1
if "Until:" in line:
nbLast += 1
if "</tr>" in line:
break
line = ""
seen = dict(warn = warnSeen, times = nbTimes, first = nbFirst, last = nbLast)
assert seen == expected, "Expected=" + str(expected) + " ≠ Seen=" + str(seen)
os.remove(mail_filename)
def whenEmailThenCheckTimes(warnAction, expected):
if os.path.exists(mail_filename):
os.remove(mail_filename)
warnAction.act(newEntry("messageW"))
warnAction.act(newEntry("messageW"))
Action._hour = 25
warnAction.act(newEntry("messageW"))
_compareEmailWithExpected(expected)
def whenSeveralDetailsModesThenOnlyOneWarn():
if os.path.exists(mail_filename):
os.remove(mail_filename)
wAction.act(newEntry("messageW"))
wAction.act(newEntry("messageW"))
wAction.act(newEntry("messageW"))
wAction.act(newEntry("messageW"))
wAction.act(newEntry("messageW"))
wActFirst.act(newEntry("messageW"))
wActFirst.act(newEntry("messageW"))
wActFirst.act(newEntry("messageW"))
wActLast.act(newEntry("messageW"))
wActLast.act(newEntry("messageW"))
wActLast.act(newEntry("messageW"))
wActFL.act(newEntry("messageW"))
wActFL.act(newEntry("messageW"))
wActFL.act(newEntry("messageW"))
wActFL.act(newEntry("messageW"))
wActNone.act(newEntry("messageW"))
wActNone.act(newEntry("messageW"))
wActNone.act(newEntry("messageW"))
Action._hour = 25
wActNone.act(newEntry("messageW"))
_compareEmailWithExpected(dict(warn = True, times = 9, first = 2, last = 2))
def whenReportThenNewSetOfMessages():
if os.path.exists(mail_filename):
os.remove(mail_filename)
Action._hour = 25
oAction.act(newEntry("message3"))
assert os.path.exists(mail_filename)
os.remove(mail_filename)
whenEmailThenCheck3Sections()
def unitTests():
whenNewDayThenReport()
whenEmailThenCheck3Sections()
whenEmailThenCheckTimes(wActFirst, dict(warn = True, times = 1, first = 1, last = 0))
whenEmailThenCheckTimes(wActLast, dict(warn = True, times = 1, first = 0, last = 1))
whenEmailThenCheckTimes(wActFL, dict(warn = True, times = 2, first = 1, last = 1))
whenEmailThenCheckTimes(wActNone, dict(warn = True, times = 0, first = 0, last = 0))
whenSeveralDetailsModesThenOnlyOneWarn()
whenReportThenNewSetOfMessages()