netcheck.py

import sys
import os
import subprocess
if sys.version_info.major == 2:
    from xmlrpclib import ServerProxy
else:
    from xmlrpc.client import ServerProxy
import socket
socket.setdefaulttimeout(10)
import transaction
import logging
from sqlalchemy import engine_from_config
from pyramid.paster import (
    get_appsettings,
    setup_logging,
    bootstrap,
    )
from ..models import DBSession
from ..models.services.conf import ConfService
from ..models.network import (
    NetConnection,
    NetLog,
    )
#
from ..models.conf import Conf
#

CONNECTED = 0
DISCONNECTED = -1
REFUSE = -2

log = logging.getLogger(__name__)


def telnet(row):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    log.debug('telnet {ip} {port}'.format(ip=row.ip, port=row.port))
    code = sock.connect_ex((row.ip, row.port))
    return code in (0, 106)


def netstat(row):
    log.debug('netcat -nap | grep {ip}:{port}'.format(ip=row.ip, port=row.port))
    c1 = ['netstat', '-nap']
    c2 = ['grep', '{ip}:{port}'.format(ip=row.ip, port=row.port)]
    p1 = subprocess.Popen(c1, stdout=subprocess.PIPE)
    p2 = subprocess.Popen(c2, stdin=p1.stdout, stdout=subprocess.PIPE)
    s = p2.communicate()[0]
    if isinstance(s, bytes):
        s = s.decode('utf-8')
    return s.find('ESTABLISHED') > -1



class NetCheck:
    def __init__(self):
        self.username = ConfService.get_val('netcheck', 'username')
        self.password = ConfService.get_val('netcheck', 'password')
        self.url = ConfService.get_val('netcheck', 'url', 'https://h2h.opensipkd.com/rpc')
        self.rpc = ServerProxy(self.url)

    def run(self):
        first_hosts = get_first_hosts()
        for name in first_hosts:
            self.ping_all(name)

    def ping(self, row):
        if row.port:
            for f in (netstat, telnet):
                if f(row):
                    self.set_last_state(row, CONNECTED)
                    return True
        log.debug('ping {}'.format(row.ip))
        c1 = ['ping', '-c1', '-w', '2', row.ip]
        c2 = ['grep', 'received']
        c3 = ['cut', '-f4', '-d', ' ']
        p1 = subprocess.Popen(c1, stdout=subprocess.PIPE)
        p2 = subprocess.Popen(c2, stdin=p1.stdout, stdout=subprocess.PIPE)
        p3 = subprocess.Popen(c3, stdin=p2.stdout, stdout=subprocess.PIPE)
        result = p3.communicate()
        s = result[0]
        if isinstance(s, bytes):
            s = s.decode('utf-8')
        s = s.strip()
        connected = s == '1'
        if connected:
            if row.port:
                self.set_last_state(row, REFUSE)
            else:
                self.set_last_state(row, CONNECTED)
        else:
            self.set_last_state(row, DISCONNECTED)
        return connected

    def ping_all(self, name):
        q = DBSession.query(NetConnection).filter_by(name=name)
        row = q.first()
        if self.ping(row):
            self.set_gateway_connected(row)
            return True
        if row.gateway:
            return self.ping_all(row.gateway)

    def set_gateway_connected(self, row):
        if not row.gateway:
            return
        q = DBSession.query(NetConnection).filter_by(name=row.gateway)
        row = q.first()
        row.status_id = CONNECTED
        DBSession.add(row)
        DBSession.flush()
        self.send_last_state(row)
        self.set_gateway_connected(row)

    def set_last_state(self, row, status_id):
        if row.status_id == status_id:
            return
        hist = NetLog()
        hist.name = row.name
        hist.ip = row.ip
        hist.port = row.port
        hist.status_id = status_id
        DBSession.add(hist)
        DBSession.flush()
        row.status_id = status_id
        row.updated = hist.created
        DBSession.add(row)
        DBSession.flush()
        self.send_last_state(row)

    def send_last_state(self, row):
        if not self.username or not row.description:
            return
        p = {'username': self.username,
             'password': self.password,
             'description': row.description,
             'ip': row.ip,
             'status_id': row.status_id}
        if row.port:
            p['port'] = row.port
        p_ = dict(p)
        p_['password'] = '...'
        log.debug('send {url} {p}'.format(url=self.url, p=p_))
        resp = self.rpc.net_status(p)
        log.debug('recv {r}'.format(r=resp))


def get_first_hosts():
    gateways = []
    q = DBSession.query(NetConnection)
    for row in q:
        if row.gateway and row.gateway not in gateways:
            gateways.append(row.gateway)
    r = []
    for row in q:
        if row.name not in gateways:
            r.append(row.name)
    return r

#
def distro():
    cmd = ['lsb_release','-a']
    c = subprocess.Popen(cmd, stdout=subprocess.PIPE)
    s = c.communicate()
    s = s[0]
    if isinstance(s, bytes):
        s = s.decode('utf-8')
    lines = s.split('\n')
    d = {}
    for line in lines:
        items = line.split('\t')
        if items[0] == 'Distributor ID:':
            d['id'] = items[1]
        elif items[0] == 'Codename:':
            d['codename'] = items[1]
        elif items[0] == 'Release:':
            d['versi'] = float(items[1])
        elif items[0] == 'Description:':
            d['keterangan'] = items[1]
    return d

def insertDistro(DBSession, get, nama):
    q = DBSession.query(Conf).filter_by(grup='distro', nama=nama)
    c = q.first()
    if not c:
        c = Conf()
        c.grup = 'distro'
        c.nama = nama
    c.nilai = get[nama]

    DBSession.add(c)
    DBSession.flush()
#

def main(argv=sys.argv):
    config_uri = argv[1]
    setup_logging(config_uri)
    settings = get_appsettings(config_uri)
    engine = engine_from_config(settings, 'sqlalchemy.')
    DBSession.configure(bind=engine)
    bootstrap(config_uri)
    nc = NetCheck()
    nc.run()
    get = distro()
    #
    insertDistro(DBSession, get, 'id')
    insertDistro(DBSession, get, 'versi')
    insertDistro(DBSession, get,  'codename')
    insertDistro(DBSession, get, 'keterangan')
    #
    transaction.commit()