challenge-checker/new_check.py

#!/usr/bin/python2

import time
import socket
import requests
import threading

from colorama import Fore, Back, Style

TIMEOUT = 0.1
SLEEP = 0

BUFF = 8192

HOST = "192.168.1.2"

TEAM_NAME = [
    "RevID.CTF",
    "Youtube",
    "Flag Earth",
    "UMN",
    "Pengen Mandi",
    "Gundar Cysec",
    "gaboet berfaedah",
    "Fake Team",
    "UNAIR FT SHL",
    "Jangan Dibaca tJOY",
]

CHALLENGE = [
    "MTP - Crypto",
    "Sophisticated Calculator - Pwn",
    "My Secure Web App - Web",
    "Save HTML Design - Web",
    "Name Generator - Pwn",
    "Web Image Search - Web",
]

EXPECTED_ANSWER = {
    1: "      !        +-+-+-+-+-+-+-+-+-+  +-+-+-+   +-+-+-+\n      ^        |S|l|a|s|h|R|o|o|t|  |C|T|F|   |3|.|0|\n     / \\       +-+-+-+-+-+-+-+-+-+  +-+-+-+   +-+-+-+\n    /___\\\n   |=   =|    +-+-+-+-+-+-+-+  +-+-+  +-+-+-+-+-+-+-+-+\n   |  B  |    |M|i|s|s|i|o|n|  |T|o|  |D|y|s|n|o|m|i|a|\n   |  A  |    +-+-+-+-+-+-+-+  +-+-+  +-+-+-+-+-+-+-+-+\n   |  L  |\n  /|  I  |\\     [G] Get Session\n / |##!##| \\    [S] Set Session\n| / ^ | ^ \\ |   [T] Generate Token\n|/ (( | )) \\|   [L] Let's Go\n    (( ))       [C] Cancel\n     ( )\n      .\n \n",
    2: "Welcome to Online Calculator v3.1\n\nOperator supported: + - * /\nex: 2 + 2\nex: 2 + a  //'a' to use last results\n\n------------------------------------\n>",
    5: ">> Your Nick: ",
}

LOCK = threading.Lock()


class Challenge(object):
    down = 0
    stat = 1

    def __init__(self, challPort, challName):
        self.name = challName
        self.port = challPort

    def set_down(self):
        if self.stat:
            self.stat = 0
            self.down = time.time()

    def set_up(self):
        self.stat = 1
        self.down = 0

    def get_downtime(self):
        return int(round(time.time() - self.down))


class Team(object):
    def __init__(self, port, teamName):
        self.name = teamName
        self.port = port
        self.challengeList = [Challenge(index, challName)
                              for index, challName in enumerate(CHALLENGE, start=1)]


def main():
    teamList = [Team(port, teamName)
                for port, teamName in enumerate(TEAM_NAME, start=1)]

    while True:
        for team in teamList:
            for chall in team.challengeList:
                #threading._start_new_thread(test_challenge, (team, chall))
                test_challenge(team, chall)
        time.sleep(SLEEP)


def success(team, chall):
    msg = now() + "[6" + str(team.port).zfill(2) + str(chall.port).zfill(2) + \
        "] " + team.name + " " + chall.name + " is UP"
    if not chall.stat:
        msg += " again total down time = " + str(chall.get_downtime()) + "s"
        with open("new_check.log", "a") as log:
            log.write(msg + "\n")
        chall.set_up()
    LOCK.acquire()
    print(Fore.GREEN + msg + Style.RESET_ALL)
    LOCK.release()


def failed(team, chall):
    chall.set_down()
    msg = now() + "[6" + str(team.port).zfill(2) + str(chall.port).zfill(2) + "] " + \
        team.name + " " + chall.name + " is DOWN for " + \
        str(chall.get_downtime()) + "s"
    with open("new_check.log", "a") as log:
        log.write(msg + "\n")
    LOCK.acquire()
    print(Fore.RED + msg + Style.RESET_ALL)
    LOCK.release()


def now():
    return time.strftime("%X ")


def test_challenge(team, chall):
    port = 60000+team.port*100+chall.port

    '''
    if chall.port == 3:
        return
    '''

    if chall.name[-3:] == "Web":
        chk = test_web(port)
    else:
        chk = test_socket(port)

    if chk:
        success(team, chall)
    else:
        failed(team, chall)


def test_socket(port):
    try:
        socket.setdefaulttimeout(TIMEOUT)
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.connect((HOST, port))

        r = ""
        t = EXPECTED_ANSWER[port % 100]
        chk = False

        while not chk:
            r += s.recv(BUFF)
            chk = r == t

        return chk
    except KeyboardInterrupt:
        exit(0)
    except:
        return False


def test_web(port):
    try:
        r = requests.get("http://"+HOST+":"+str(port), timeout=TIMEOUT)
        chk1 = True  # r.text == EXPECTED_ANSWER[get_chall(port)]
        chk2 = r.status_code == requests.codes.ok
        if chk1 and chk2:
            return True
        else:
            return False
    except KeyboardInterrupt:
        exit(0)
    except:
        return False


if __name__ == '__main__':
    main()