tcp check

tcp-chcker.py

#!/usr/bin/env python
# encoding: utf-8

import socket
import struct
import select
import errno
import time

# https://github.com/tevino/tcp-shaker
# https://yanagieiichi.github.io/tcp-check/docs/intro


class TCPChecker(object):

    def __init__(self):
        self.epoll = select.epoll()
        self.bucket = dict()

    def create_socket(self):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.setsockopt(socket.IPPROTO_TCP, socket.TCP_QUICKACK, 0)
        s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 0)
        s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
        s.setblocking(False)
        return s

    def connect(self, s, ip, port, timeout):
        start = time.time()
        while True:
            if time.time() - start > 1:
                return False
            try:
                s.connect((ip, port))
                return True
            except OSError as err:
                if err.errno in [errno.EINPROGRESS, errno.EALREADY, errno.EINTR]:
                    continue
                return False

    def down(self, ip, port):
        s = self.create_socket()
        if not self.connect(s, ip, port, 0.5):
            return False
        self.epoll.register(s.fileno(), select.EPOLLOUT | select.EPOLLET)
        while True:
            events = self.epoll.poll(1)
            for fileno, event in events:
                if fileno == s.fileno():
                    ret = s.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
                    print("out", ret)
                    if ret == 0:
                        s.close()
                        return True
                    return False

if __name__ == '__main__':
    checker = TCPChecker()
    print(checker.down('123.125.115.110', 80))