#!/usr/bin/env python3 # $Id: hbd,v 1.38 2013/07/14 02:25:05 andreas Exp $ # Wait for heartbeat messages and act on them (or their absence) # VER = 4.2 import time import os import string import sys import socket import ssl import pathlib import atexit import select import socketserver import http.server import getopt import signal import pickle import smtplib import traceback import urllib.request, urllib.parse, urllib.error import urllib.parse import http.client import threading import subprocess from hashlib import md5 import json import zlib import codecs import asyncio import websockets from subprocess import Popen, STDOUT, PIPE #from hbdclass import * import hbdclass CERT_PATH="/usr/local/etc/letsencrypt/live/w02.wrede.ca/" WSS_PEM = CERT_PATH + "fullchain.pem" WSS_KEY = CERT_PATH + "privkey.pem" NSUPDATE_BIN = "/usr/local/bin/nsupdate" # override in .hbrc possible SEND_EMAIL=False SEND_PUSHOVER=True DEBUG = 0 hbdclass.DEBUG = DEBUG MAXRECV = 32767 LOGFILE = "/home/andreas/public_html/messages/andreas" PICKFILE = "/var/tmp/hbd.pick" AEMAIL = ["andreas@wrede.ca"] NAME = "heatbeat" SMTPSERVER = "localhost" msgs = [] #AEW upcount = 0 PORT = 50003 TPORT = 50004 THOST = "" WSPORT = 50005 WSSPORT = 50006 verbose = False INTERVAL = 10 GRACE = 2 DROPOVERDUE = 7*24*3600 os.environ['TZ'] = 'EST5EDT' tsfm=["%H","%d","%U"] lastfm=["","",""] tcss = """ """ def handler(signum, frame): global running, sig sig = signum if not running: if verbose: sys.stderr.write("NOT runing signal: %s running: %d" % (sig, running)) sys.exit(2) if verbose: sys.stderr.write("signal: %s running: %s frame: %s" % (sig, running, frame)) def shortname(name): r = name.split('.') return r[0] class NullDevice: def write(self, s): pass class LogDevice: def __init__(self): self.fh = open("/tmp/log1","a") def write(self, s): self.fh.write(s) self.fh.flush() def dicttos(ID, d, compress=False): s = [] for k in d: if type(d[k]) == type(1.2): s.append("%s=%0.5f" % (k, d[k])) else: s.append("%s=%s" % (k, d[k])) pk = ";".join(s) if compress: zpk = zlib.compress(pk.encode(), 6) ID = "!" + ID + ":" opk = ID.encode() + zpk else: zpk = pk opk = ID + ":" + zpk return opk def stodict(msg): d = {} if len(msg) > 0 and chr(msg[0]) == "!": pk = zlib.decompress(msg[5:]).decode() d['ID'] = msg[1:4].decode() else: r0 = msg.split(':',1) pk = r0[1] d['ID'] = r0[0] r = pk.split(';') for v in r: vr = v.split('=', 1) k = vr[0].strip() if len(vr) == 1: d[k] = None else: v = vr[1].strip() if v[0].isdigit(): v = eval(v) d[k] = v return d def oldmtodict(msg): return stodict('HTB:'+msg) def email(s, msg): if not SEND_EMAIL: return ret = "OK" toaddrs = AEMAIL fromemail = "aew.heartbeat@wrede.ca" subj = "Info from %s: %s" % (NAME, s) date = time.strftime("%a, %d %b %Y %H:%M:%S %z", time.localtime()) body = "To: %s\nFrom: %s\nSubject: %s\nDate: %s\n\n%s" % (toaddrs[0], fromemail, subj, date, msg) try: server = smtplib.SMTP(SMTPSERVER) if DEBUG > 0: server.set_debuglevel(1) server.sendmail(fromemail, toaddrs, body) except smtplib.SMTPRecipientsRefused as errs: log(None, "cannot send email: %s\n" % (errs)) ret = "Fail" except: print(("smtp error: "+traceback.format_exc())) saveandrestart() try: server.quit() except: pass return ret def pushmsg(msg): if pushsrv in ["all", "pushover"]: pushover(msg) if pushsrv in ["all","mattermost"]: pushmattermost(msg) if pushsrv in ["all","signal"]: pushsignal(msg) if pushsrv in ["all"]: print("notice:", msg) def pushover(msg): if not SEND_PUSHOVER: return conn = http.client.HTTPSConnection("api.pushover.net:443") try: conn.request("POST", "/1/messages.json", urllib.parse.urlencode({ "token": "ac7NLX2rPjXFareeDgLpXNoDf4iFmf", "user": "uDhH33UjQQDYtNzJb1ThRiWb9ingGK", "message": msg, }), { "Content-type": "application/x-www-form-urlencoded" }) conn.getresponse() except: pass CHANNEL = "Monitoring" TOKEN = "rxz6b3886iygxnhbzpmgbsrocy" HOST = "192.168.10.101" ICON = "https://in-transit.ca/HeartBeat.png" USERNAME = "admin" def pushmattermost(msg): ses = { 'url': HOST, 'scheme':'http', 'basepath': '/api/v4', 'port':8065, } mm = Driver(ses) msg = { "text": msg, "channel": CHANNEL, "username": USERNAME, "icon_url": ICON } try: rc = mm.webhooks.call_webhook(TOKEN, msg) except Exception as e: rc = str(e) if not rc: print(rc) USER = "+16472472447" RECIPIENT = "+14168226179" def pushsignal(msg, title="hbd", recipient=RECIPIENT): message = f'"{title}: {msg}"' CLI = [ "/usr/bin/ssh", "andreas@w02", "/usr/local/bin/signal-cli", "-u", USER, "send", "-m", message, # "-g", GROUP, recipient, ] if verbose: print(f"DBG cli: {CLI}") res = subprocess.run(CLI, shell=False, capture_output=True) rc = res.returncode == 0 print(res.stdout.decode()) if not rc: print(f"signalcli failed: {res.stderr.decode()}") else: if verbose: print(f"signalcli msg sent, res {res.stdout.decode()}") return rc # nsupdate: set the DNS A record for a fqdn # return: None if ok, else error text def nsupdate(hostname, newip, dyndomain): D = {} D['domain'] = dyndomain D['fqdn'] = "%s.dy.%s" % (hostname, dyndomain) D['dnsttl'] = '5' D['newip'] = newip D['ts'] = time.strftime('%Y-%m-%d.%H:%M:%S', time.gmtime()) if newip.find(":") > 0: nsup = """update delete %(fqdn)s AAAA update add %(fqdn)s %(dnsttl)s AAAA %(newip)s update delete %(fqdn)s TXT update add %(fqdn)s %(dnsttl)s TXT "Created: %(ts)s" send answer """ % D else: nsup = """update delete %(fqdn)s A update add %(fqdn)s %(dnsttl)s A %(newip)s update delete %(fqdn)s TXT update add %(fqdn)s %(dnsttl)s TXT "Created: %(ts)s" send answer """ % D if DEBUG > 0: log(None, "DBG: nsup %s" % nsup) cmd = [nsupdate_bin, "-k", "/etc/dhcpc/Kdy.%(domain)s.+157+00000." % D, "-v"] if DEBUG > 0: log(None, "DBG: cmd %s" % cmd) try: p = Popen(cmd, shell=False, bufsize=1, stdin=PIPE, stdout=PIPE, stderr=STDOUT) except OSError as e: return "nsupdate: execution failed: %s" % e except: return "nsupdate: some error occured" (output, err) = p.communicate(nsup.encode()) if output.decode().find('status: NOERROR') >= 0: return None return output.decode()+err.decode() # def dur(sec): sec = int(sec) h = int(sec / 3600) m = int((sec - h * 3600) / 60) s = int((sec - h * 3600) % 60) if h > 0: return "%d:%02d:%02d" % (h, m, s) if m > 0: return "%d:%02d" % (m, s) return "0:%02d" % s def fixsort(): s = list(hbdclass.Host.hosts.keys()) s.sort() x = 0 for n in s: hbdclass.Host.hosts[n].num = x x += 1 # def on_exit(): if DEBUG > 0: sys.stderr.write("on_exit\n") try: logf.close() except: pass print("exit") def initlog(logfile): try: return open(logfile, "a+") except: pass try: return open(logfile, "w") except Exception as e: print("cannot open loffile %s, using STDERR: %s" % (logfile, e)) return sys.stderr # # def checkoverdue(): now = time.time() for h in list(hbdclass.Host.hosts.keys()): pmsg = [] for c in hbdclass.Host.hosts[h].connections: conn = hbdclass.Host.hosts[h].connections[c] if conn.state == hbdclass.Connection.down: continue timeout = hbdclass.Host.hosts[h].interval + grace if conn.state == hbdclass.Connection.up and (now - conn.lastbeat) > timeout: conn.newstate(hbdclass.Connection.overdue, now, grace) pmsg.append(conn.afam) if conn.state == hbdclass.Connection.overdue and (now - conn.lastbeat) > DROPOVERDUE: conn.newstate(hbdclass.Connection.unknown, conn.lastbeat) if pmsg != []: if h in watchhosts: email("overdue", "%s overdue" % " and ".join(pmsg)) pushmsg("%s %s overdue" % (h, " and ".join(pmsg))) log(h, "%s overdue" % " and ".join(pmsg)) msg_to_websockets('host', hbdclass.Host.hosts[h].stateinfo()) def log(host, m, service=None): if DEBUG > 0: print("Log: %s %s" % (host, m)) now = time.time() ts = time.strftime("%b %d %H:%M:%S", time.localtime(now)) if service: srv = "service %s: " % service else: srv = "" if host: hst = "%s " % host else: hst = "" msg = "%s: %s%s%s" % (ts, hst, srv, m) msgs.append(msg+'\n') msg_to_websockets('message', msg) if logfmt == "msg": m2 = "%d|%s|%s\n" % (now, hst, m) else: m2 = msg+'\n' logf.write(m2) logf.flush() pickleit() def dnsupdatethread(): while True: name, addr = hbdclass.Host.dnsQ.get() m = "changed address to %s" % (addr) for dyndomain in dyndomains: err = nsupdate(name, addr, dyndomain) if err: m += ", DNS update failed: %s" % err email("error: nsupdate failed", "%s.dy.%s: %s" % (name, dyndomain, m)) else: m += ", DNS updated." hbdclass.Host.dnsQ.task_done() log(name, m) # # # # def readsock(sock): global now if DEBUG > 3: sys.stderr.write("readsock recfrom start") now = time.time() data, addrp = sock.recvfrom(MAXRECV) if DEBUG > 3: sys.stderr.write("readsock = %s, %s\n" % (data,addrp)) try: msg = stodict(data) except: return if DEBUG > 3: sys.stderr.write("msg is %s" % str(msg)) if not msg: # Old hbc client if verbose: print(("old hbc:", data)) msg = oldmtodict(data) if DEBUG > 2: print(("readsock = %s, %s" % (msg,addrp))) addr = addrp[0:2] name = shortname(msg.get('name', "unknown")) if not name in hbdclass.Host.hosts: # was: hosts.has_key(name): host = hbdclass.Host(name) host.dyn = name in dyndnshosts if verbose: print(("XX: New host, num now %s" % (len(hbdclass.Host.hosts)))) newh=True else: host = hbdclass.Host.hosts[name] newh=False cid = msg.get('id', 0) try: rtt = float(msg.get('rtt',None)) except: rtt = None if msg['ID'] == 'HTB': host.doesack = msg.get('acks', -1) host.setcver(msg.get('ver', 0)) interval = int(msg.get('interval', 0)) shutdown = msg.get('shutdown', 0) service = msg.get('service', "unknown") message = msg.get('msg', None) boot = msg.get('boot', 0) conn, res = host.conndata(cid, addr[0], rtt, now) if res: log(name, res) if name in watchhosts: email("address change", "%s %s" % (host.name, res)) pushmsg("%s %s" % (host.name, res)) if boot: log(name, "booted") if name in watchhosts: m = "%s booted" % (host.name) email("booted", m) pushmsg(m) if message: log(name, "msg: %s" % message, service=service) if name in watchhosts: email("msg", message) pushmsg(message) if conn.getstate() != hbdclass.Connection.up: # XXX and interval > 0: lasts = conn.state d = conn.newstate(hbdclass.Connection.up, now) m = "%s back after being %s for %s" % (conn.afam, lasts, dur(d)) log(name, m) if name in watchhosts: email("%s back" % conn.afam, name) pushmsg("%s %s is back" % (name, conn.afam)) if boot or newh: host.upcount = host.doesack else: host.upcount += 1 if shutdown: log(name, "%s shutdown" % conn.afam) if name in watchhosts: email("shutdown", "%s %s shutdown" % (name, conn.afam)) pushmsg("%s %s shutdown" % (name, conn.afam)) conn.newstate(hbdclass.Connection.down, now) if interval > 0: host.interval = interval rmsg = {'time': time.time()} op = 'ACK' if host.cver < 1: opkt = 'ACK' rmsg = 'ACK' else: opkt = dicttos('ACK', rmsg, host.cver > 1) # clients w/ ver 2+ can cope try: ss=sock.sendto(opkt, addr) except: pass # XXX return pkg failes if DEBUG > 2: print(("sendto1: %s (%s) %s %s" % (addr, len(opkt), op, str(rmsg)[:50]))) # send any commands we have queued while len(host.cmds): op, rmsg = host.cmds[0] if op == 'CMD': email("%s cmd exec" % name, "command '%s' sent" % rmsg) del host.cmds[0] log(name, "command sent") if host.cver < 1: rmsg = rmsg['cmd'] elif op == 'UPD': del host.cmds[0] log(name, "update initiated") if host.cver < 1: log(name," ver 0 does not support UPD") continue if host.cver < 1: opkt = rmsg op = "" else: opkt = dicttos(op, rmsg, True) try: ss=sock.sendto(opkt, addr) except Exception as e: print(("opkt len is %s" % len(opkt))) print(("cannot send: %s" % e)) if verbose: print(("sendto2: %s (%s) %s %s" % (addr, len(opkt), op, str(rmsg)[:50]))) if DEBUG > 2: print(("msg from %s,%s, sent %s bytes back" % (addr[0], addr[1], ss))) msg_to_websockets('host', host.stateinfo()) def updatecode(ucode, uname): fail = None try: fh = open(ucode, "r") new_code = fh.read() fh.close() except Exception as e: fail = "cannot read new code: %s" % e if not fail: m = md5() new_codeE = new_code.encode() m.update(new_codeE) icsum = m.hexdigest() rmsg = {'csum': icsum, 'code': codecs.encode(new_codeE, 'base64') } hbdclass.Host.hosts[uname].cmds.append(('UPD',rmsg)) return fail # # Web Server # class HttpServer(socketserver.ThreadingMixIn, http.server.HTTPServer): allow_reuse_address = True def threaded(self): pass # # class HttpHandler(http.server.BaseHTTPRequestHandler): server_version = "HeartbeatHTTP/%s" % VER def version_string(self): return self.server_version def handle(self): # return http.server.BaseHTTPRequestHandler.handle(self) try: return http.server.BaseHTTPRequestHandler.handle(self) except Exception as e: self.log_error("Request went away: %r", e) self.close_connection = 1 return def do_HEAD(self): self.setheaders(200) def setheaders(self, code, headerdict={}): self.send_response(code) self.send_header("Last-Modified", time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(now))) # self.send_header("Accept-Ranges","bytes") # self.send_header("hbdclass.Connection","close") for h in headerdict: self.send_header(h, headerdict[h]) self.end_headers() def buildhead(self, title="Heartbeat", refresh=None, extras=None): res=[] res.append('') res.append("") res.append("
") res.append('%s (%s)
' % (time.strftime("%H:%M:%S", time.localtime(now)), os.environ.get('TZ', 'CET-1CDT'))) res.append("") return res def builderror(self, code, cause, lcause): res=[] res.append('') res.append('') res.append('%s
' % lcause) res.append('