tsg-tsg-diagnose/unittest_python/unittest/unittest_self.py

import unittest
import json
import pycurl
import os
import re
import time
from io import BytesIO


URLBypass         = 'https://sha384.badssl.self-test.geedge.net'
URLIntercept      = 'https://sha256.badssl.self-test.geedge.net'
URLSexpired       = 'https://expired.badssl.self-test.geedge.net'
URLSwronghost     = 'https://wrong.host.badssl.self-test.geedge.net'
URLSselfsigned    = 'https://self-signed.badssl.self-test.geedge.net'
URLSuntrustedroot = 'https://untrusted-root.badssl.self-test.geedge.net'
URLSrevoked       = 'https://revoked.badssl.self-test.geedge.net'
URLSpinningtest   = 'https://pinning-test.badssl.self-test.geedge.net'

URLRedirect       = 'https://cn.bing.com/rs/2V/pE/cj,nj/b1392357/d94c45f4.js'
URLReplace        = 'https://cn.bing.com/rs/5j/1pF/cj,nj/2213d9b6/b50738ca.js'
URLInsert         = 'https://cn.bing.com/?FORM=BEHPTB'
URLHijack         = 'https://cn.bing.com/rs/31/2n/cj,nj/4c7364c5/40e1b425.js'
URLBlock          = 'https://cn.bing.com/rs/31/22/cj,nj/3f1e2270/f8c6dd44.js'


ssl_bypass_info_re           = "ssl bypass ok"
ssl_intercept_info_re        = "ssl intercept ok"
https_exprired_info_re       = "https exprired ok"
https_wrong_host_info_re     = "https wrong host ok"
https_self_signed_info_re    = "https self signed ok"
https_untrusted_root_info_re = "https untrusted_root ok"
https_revoked_info_re        = "https revoked ok"
https_pinning_test_info_re   = "https pinning-test ok"
http_redirect_info_re        = "http connection redirect success"
http_replace_info_re         = "http connection replace success"
http_insert_info_re          = "http connection insert success"
http_hijack_info_re          = "http connection hijack success"
http_block_info_re           = "http connection block success"

wpr_dns_resolve  = ['cn.bing.com:443:192.0.2.131']

class SSLCheckRequestBuild:
    def __init__(self):
        self.conn = pycurl.Curl()
        self.conn.setopt(self.conn.WRITEFUNCTION, BytesIO().write)
        self.conn.setopt(self.conn.OPT_CERTINFO, 1)
        self.conn.setopt(self.conn.SSL_VERIFYPEER, False)

    def ssl_bypass(self):
        self.conn.setopt(self.conn.URL,URLBypass)
        self.conn.perform()
        certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
        self.conn.close()
        issuer = ()
        for cert_info in certs[0]:
            if cert_info[0] == "Issuer":
                issuer = cert_info
                break
        if len(issuer) <= 0:
            raise Exception("Error: get Certificate info error")
        if re.search(r'\bCN[\s]*=[\s]*BadSSL\b',issuer[1],0):
            raise Exception(ssl_bypass_info_re)
        elif re.search(r'\bCN = Tango[\s\S]*UNTRUST\b',issuer[1],0):
            raise Exception("Error:Ssl link is intercepted")
        else:
            raise Exception("Error:Got other error certificate information")

    def ssl_intercept(self):
        self.conn.setopt(self.conn.URL,URLIntercept)
        self.conn.perform()
        certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
        self.conn.close()
        issuer = ()
        for cert_info in certs[0]:
            if cert_info[0] == "Issuer":
                issuer = cert_info
                break
        if len(issuer) <= 0:
            raise Exception("Error: get Certificate info error")
        if re.search(r'CN = Tango[\s\S]*UNTRUST',issuer[1],0):
            raise Exception(ssl_intercept_info_re)
        elif re.search(r'\bCN[\s]*=[\s]*BadSSL\b',issuer[1],0):
            raise Exception("Error:Ssl link is intercepted")
        else:
            raise Exception("Error:Got other error certificate information")


class SslInterceptRequestBuild:
    def __init__(self):
        self.conn = pycurl.Curl()
        self.conn.setopt(self.conn.WRITEFUNCTION, BytesIO().write)
        self.conn.setopt(self.conn.OPT_CERTINFO, 1)
        self.conn.setopt(self.conn.SSL_VERIFYPEER, False)

    def ssl_intercept_expired(self):
        self.conn.setopt(self.conn.URL, URLSexpired)
        self.conn.perform()
        certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
        self.conn.close()
        issuer = ()
        for cert_info in certs[0]:
            if cert_info[0].lower() == "issuer":
                issuer = cert_info
                break
        if len(issuer) <= 0:
            raise Exception("Error: get Certificate info error")
        if re.search(r'\bCN[\s]*=[\s]*Tango\b',issuer[1],0):
            if re.search(r'\bCN = Tango[\s\S]*UNTRUST\b',issuer[1],0):
                raise Exception(https_exprired_info_re)
            else:
                raise Exception("Fail:ssl intercept cert is trust")
        else:
            raise Exception("Error:Got other error certificate information")


    def ssl_intercept_wrong_host(self):
        self.conn.setopt(self.conn.URL,URLSwronghost)
        self.conn.setopt(self.conn.SSL_VERIFYHOST, False)
        self.conn.perform()
        certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
        self.conn.close()
        issuer = ()
        for cert_info in certs[0]:
            if cert_info[0].lower() == "issuer":
                issuer = cert_info
                break
        if len(issuer) <= 0:
            raise Exception("Error: get Certificate info error")
        if re.search(r'\bCN[\s]*=[\s]*Tango\b',issuer[1],0):
            if re.search(r'\bCN = Tango[\s\S]*UNTRUST\b',issuer[1],0):
                raise Exception(https_wrong_host_info_re )
            else:
                raise Exception("Fail:ssl intercept cert is trust")
        else:
            raise Exception("Error:Got other error certificate information")

    def ssl_intercept_self_signed(self):
        self.conn.setopt(self.conn.URL,URLSselfsigned)
        self.conn.perform()
        certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
        self.conn.close()
        issuer = ()
        for cert_info in certs[0]:
            if cert_info[0].lower() == "issuer":
                issuer = cert_info
                break
        if len(issuer) <= 0:
            raise Exception("Error: get Certificate info error")
        if re.search(r'\bCN[\s]*=[\s]*Tango\b',issuer[1],0):
            if re.search(r'\bCN = Tango[\s\S]*UNTRUST\b',issuer[1],0):
                raise Exception(https_self_signed_info_re)
            else:
                raise Exception("Fail:ssl intercept cert is trust")
        else:
            raise Exception("Error:Got other error certificate information")

    def ssl_intercept_untrusted_root(self):
        self.conn.setopt(self.conn.URL,URLSuntrustedroot)
        self.conn.perform()
        certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
        self.conn.close()
        issuer = ()
        for cert_info in certs[0]:
            if cert_info[0].lower() == "issuer":
                issuer = cert_info
                break
        if len(issuer) <= 0:
            raise Exception("Error: get Certificate info error")
        if re.search(r'\bCN[\s]*=[\s]*Tango\b',issuer[1],0):
            if re.search(r'\bCN = Tango[\s\S]*UNTRUST\b',issuer[1],0):
                raise Exception(https_untrusted_root_info_re)
            else:
                raise Exception("Fail:ssl intercept cert is trust")
        else:
            raise Exception("Error:Got other error certificate information")


    def ssl_intercept_revoked(self):

        self.conn.setopt(self.conn.URL,URLSrevoked)
        self.conn.perform()
        certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
        self.conn.close()
        issuer = ()
        for cert_info in certs[0]:
            if cert_info[0].lower() == "issuer":
                issuer = cert_info
                break
        if len(issuer) <= 0:
            raise Exception("Error: get Certificate info error")
        if re.search(r'\bCN[\s]*=[\s]*Tango\b',issuer[1],0):
            if re.search(r'\bCN = Tango[\s\S]*UNTRUST\b',issuer[1],0):
                raise Exception(https_revoked_info_re)
            else:
                raise Exception("Fail:ssl intercept cert is trust")
        else:
            raise Exception("Error:Got other error certificate information")


    def ssl_intercept_pinning_test(self):

        self.conn.setopt(self.conn.URL,URLSpinningtest)
        self.conn.perform()
        certs = self.conn.getinfo(self.conn.INFO_CERTINFO)
        self.conn.close()
        issuer = ()
        for cert_info in certs[0]:
            if cert_info[0].lower() == "issuer":
                issuer = cert_info
                break
        if len(issuer) <= 0:
            raise Exception("Error: get Certificate info error")
        if re.search(r'\bCN[\s]*=[\s]*Tango\b',issuer[1],0):
            if re.search(r'\bCN = Tango[\s\S]*UNTRUST\b',issuer[1],0):
                raise Exception(https_pinning_test_info_re)
            else:
                raise Exception("Fail:ssl intercept cert is trust")
        else:
            raise Exception("Error:Got other error certificate information")


class SslHttpRequestBuild:
    def __init__(self):
        self.bodyBuf = BytesIO()
        self.conn = pycurl.Curl()
        self.conn.setopt(self.conn.OPT_CERTINFO, 1)
        self.conn.setopt(self.conn.SSL_VERIFYPEER, False)
        self.conn.setopt(self.conn.ENCODING, "gzip,deflate")
        self.conn.setopt(self.conn.RESOLVE,wpr_dns_resolve)

    def http_redirect(self):
        self.conn.setopt(self.conn.URL, URLRedirect)
        self.conn.setopt(self.conn.WRITEFUNCTION, BytesIO().write)
        self.conn.perform()
        rescode = self.conn.getinfo(self.conn.RESPONSE_CODE)
        self.conn.close()
        if rescode == 301 or rescode == 302:
            raise Exception(http_redirect_info_re)
        else:
            raise Exception("Error:Http connection redirect fail")

    def http_replace(self):
        self.conn.setopt(self.conn.WRITEDATA, self.bodyBuf)
        self.conn.setopt(self.conn.URL, URLReplace)
        resCode = self.conn.getinfo(self.conn.RESPONSE_CODE)
        self.conn.perform()
        body = self.bodyBuf.getvalue().decode('utf-8')
        self.conn.close()
        if not re.search(r'EnglishSearchShared', body, 0) and \
                re.search(r'03C174CD9D809789CCEC18D6F585DF3E', body, 0):
            raise Exception(http_replace_info_re)
        else:
            raise Exception("Error:Http connection replace fail")

    def http_insert(self):
        self.conn.setopt(self.conn.WRITEDATA, self.bodyBuf)
        self.conn.setopt(self.conn.URL, URLInsert)
        resCode = self.conn.getinfo(self.conn.RESPONSE_CODE)
        self.conn.perform()
        body = self.bodyBuf.getvalue().decode('utf-8')
        self.conn.close()
        if re.search(r'httpSelfcheckInsert', body, 0) and \
                re.search(r'5BE3754D1EA8D51E8D993060FA225330', body, 0):
            raise Exception(http_insert_info_re)
        else:
            raise Exception("Error:Http connection insert fail")

    def http_block(self):
        self.conn.setopt(self.conn.URL, URLBlock)
        self.conn.setopt(self.conn.WRITEFUNCTION, self.bodyBuf.write)
        self.conn.perform()
        rescode = self.conn.getinfo(self.conn.RESPONSE_CODE)
        body = self.bodyBuf.getvalue().decode('utf-8')
        self.conn.close()
        if re.search(r'E33F01E50AFE043191931DD40190B09B', body, 0) and (rescode == 404 or rescode == 451):
            raise Exception(http_block_info_re)
        else:
            raise Exception("Error:http connection block fail")

    def http_hijack(self):

        self.conn.setopt(self.conn.URL, URLHijack)
        self.conn.setopt(self.conn.WRITEFUNCTION, self.bodyBuf.write)
        self.conn.perform()
        self.conn.close()
        if os.path.exists("/root/http_hijack.out"):
            os.remove("/root/http_hijack.out")
        cmdtodo = 'curl %s -k -s --resolve cn.bing.com:443:192.0.2.131 -o /root/http_hijack.out' % URLHijack
        optdl = os.popen(cmdtodo)
        if len(optdl.read()):
            optdl.close()
            raise Exception("Error:http_hijack download file fail")
        optdl.close()
        if not os.path.exists("/root/http_hijack.out"):
            raise Exception("Error:http_hijack download file fail")
        optmd5 = os.popen("md5sum /root/http_hijack.out")
        if re.search("4bf06db1a228c5c8d978ebf9e1169d0d", optmd5.read(), 0):
            optmd5.close()
            raise Exception(http_hijack_info_re)
        else:
            optmd5.close()
            raise Exception("Error:http connection hijack fail")


class SslUnitTest(unittest.TestCase):

    def test_ssl_bypass(self):
        sslHandler = SSLCheckRequestBuild()
        with self.assertRaisesRegex(Exception, ssl_bypass_info_re):
            sslHandler.ssl_bypass()

    def test_ssl_intercept(self):
        sslHandler = SSLCheckRequestBuild()
        with self.assertRaisesRegex(Exception, ssl_intercept_info_re):
            sslHandler.ssl_intercept()

    def test_ssl_intercept_expired(self):
        requestHandler = SslInterceptRequestBuild()
        with self.assertRaisesRegex(Exception, https_exprired_info_re):
            requestHandler.ssl_intercept_expired()

    def test_ssl_intercept_wrong_host(self):
        requestHandler = SslInterceptRequestBuild()
        with self.assertRaisesRegex(Exception, https_wrong_host_info_re):
            requestHandler.ssl_intercept_wrong_host()

    def test_ssl_intercept_self_signed(self):
        requestHandler = SslInterceptRequestBuild()
        with self.assertRaisesRegex(Exception, https_self_signed_info_re):
            requestHandler.ssl_intercept_self_signed()

    def test_ssl_intercept_untrusted_root(self):
        requestHandler = SslInterceptRequestBuild()
        with self.assertRaisesRegex(Exception, https_untrusted_root_info_re):
            requestHandler.ssl_intercept_untrusted_root()

    def test_ssl_intercept_revoked(self):
        requestHandler = SslInterceptRequestBuild()
        with self.assertRaisesRegex(Exception, https_revoked_info_re):
            requestHandler.ssl_intercept_revoked()

    def test_ssl_intercept_pinning_test(self):
        requestHandler = SslInterceptRequestBuild()
        with self.assertRaisesRegex(Exception, https_pinning_test_info_re):
            requestHandler.ssl_intercept_pinning_test()

    def test_http_redirect(self):
        httpHandler = SslHttpRequestBuild()
        with self.assertRaisesRegex(Exception, http_redirect_info_re):
            httpHandler.http_redirect()

    def test_http_block(self):
        httpHandler = SslHttpRequestBuild()
        with self.assertRaisesRegex(Exception, http_block_info_re):
            httpHandler.http_block()

    def test_http_replace(self):
        httpHandler = SslHttpRequestBuild()
        with self.assertRaisesRegex(Exception, http_replace_info_re):
            httpHandler.http_replace()

    def test_http_hijack(self):
        httpHandler = SslHttpRequestBuild()
        with self.assertRaisesRegex(Exception, http_hijack_info_re):
            httpHandler.http_hijack()

    def test_http_insert(self):
        httpHandler = SslHttpRequestBuild()
        with self.assertRaisesRegex(Exception, http_insert_info_re):
            httpHandler.http_insert()

if __name__ == '__main__':
    suite = unittest.TestSuite()
    suite._cleanup = False
    suite.addTest(SslUnitTest('test_ssl_bypass'))
    suite.addTest(SslUnitTest('test_ssl_intercept'))
    suite.addTest(SslUnitTest('test_ssl_intercept_expired'))
    suite.addTest(SslUnitTest('test_ssl_intercept_wrong_host'))
    suite.addTest(SslUnitTest('test_ssl_intercept_self_signed'))
    suite.addTest(SslUnitTest('test_ssl_intercept_untrusted_root'))
    suite.addTest(SslUnitTest('test_ssl_intercept_revoked'))
    suite.addTest(SslUnitTest('test_ssl_intercept_pinning_test'))
    suite.addTest(SslUnitTest('test_http_redirect'))
    suite.addTest(SslUnitTest('test_http_block'))
    suite.addTest(SslUnitTest('test_http_replace'))
    suite.addTest(SslUnitTest('test_http_hijack'))
    suite.addTest(SslUnitTest('test_http_insert'))
    while True:
        logpath = "/root/result_self_test/unittest/unittest_result.log." + time.strftime("%Y-%m-%d",time.localtime())
        with open(logpath,"w+") as f:
            runner = unittest.TextTestRunner(stream=f,verbosity=2)
            f.write("Test start time: ")
            f.write(time.asctime(time.localtime(time.time())))
            f.write("\n")
            runner.run(suite)
        time.sleep(1)