gfwleak/tsg-tsg-diagnose
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 1. 添加测试dns防火墙的测试例

Browse Source 
2. 修改dns的openfile的镜像源, 解决源镜像无法下载问题
This commit is contained in:
yangwenlin
2021-08-01 23:27:55 -07:00
committed by 付明卫
parent fddc32b5f2
commit fec3b564b8
12 changed files with 482 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

270
unittest_python/unittest/tsg_diagnose.py
View File

@@ -13,9 +13,20 @@ from telegraf.client import TelegrafClient
import hashlib
from configparser import ConfigParser
import random
import dns.exception
import dns.resolver
import sys
suite_test_config_dict = {'test_securityPolicy_bypass': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
'test_dnsRequest_deny_drop': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
'test_dnsRequest_deny_redirect_a': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
'test_dnsRequest_deny_redirect_aaaa': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
'test_dnsRequest_deny_redirect_a_range_ttl': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
'test_dnsRequest_deny_redirect_aaaa_range_ttl': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
'test_dnsRequest_allow_rdtype_a': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
'test_dnsRequest_allow_rdtype_aaaa': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
'test_dnsRequest_allow_rdtype_cname': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
'test_securityPolicy_intercept': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
'test_securityPolicy_intercept_certerrExpired': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
'test_securityPolicy_intercept_certerrSelf_signed': {'enabled':1,'conn_timeout':1,'max_recv_speed_large':6553600},
@@ -123,6 +134,215 @@ URLSslFirewallAllow = "https://sha512.badssl.self-test.geedge.net"
URLSslFirewallDenyDrop = "https://rsa2048.badssl.self-test.geedge.net"
URLSslFirewallDenyRst = "https://rsa4096.badssl.self-test.geedge.net"
DNS_SERVER_ALLOW_TTL = 60
DNS_SERVER_REDIRECT_TTL = 333
DNS_SERVER_REDIRECT_RANGE_LOW = 400
DNS_SERVER_REDIRECT_RANGE_HIGH = 500
DNS_SERVER_IP = ["192.0.2.135"]
DnsRequestFirewallDenyDrop = "Dns request timeout is deny drop sucess"
DnsARequestFireWallDenyRedirect = "Dns rdtype A request is deny reidrect sucess"
DnsAAAARequestFireWallDenyRedirect = "Dns rdtype AAAA request is deny redirect sucess"
DnsARequestFireWallDenyRedirectRangTTL = "Dns rdtype A request is deny reidrect and range ttl sucess"
DnsAAAARequestFireWallDenyRedirectRangTTL = "Dns rdtype AAAA request is deny redirect and range ttl sucess"
DnsARequestFirewallAllow = "Dns rdtype A request data is sucess"
DnsAAAARequestFirewallAllow = "Dns rdtype AAAA request data is sucess"
DnsCNAMERequestFirewallAllow = "Dns rdtype CNAME request data is sucess"
class DNSCheckRequestBuild:
def dns_action_deny_subaction_drop(self):
dns_resolver=dns.resolver.Resolver()
dns_resolver.nameservers = DNS_SERVER_IP
dns_resolver.timeout = float(3)
dns_resolver.lifetime = float(3)
try:
dns_answer = dns_resolver.query("www.3test-ipv4.com", 'A')
except dns.exception.DNSException as errorinfo:
if type(errorinfo) == dns.exception.Timeout:
raise Exception(DnsRequestFirewallDenyDrop)
else:
raise Exception("Error: The dns_action_deny_subaction_drop check failure, code: %s" % errorinfo)
else:
raise Exception("Error: The dns_action_deny_subaction_drop test deny drop failure" )
def dns_action_deny_subaction_redirect_a(self):
dns_resolver=dns.resolver.Resolver()
dns_resolver.nameservers = DNS_SERVER_IP
dns_resolver.timeout = float(3)
dns_resolver.lifetime = float(3)
try:
dns_answer = dns_resolver.query("www.2test-ipv4.com", 'A')
except dns.exception.DNSException as errorinfo:
raise Exception("Error: The dns_action_deny_subaction_redirect_a check failure, code: %s" % errorinfo)
else: # drop-redirect and respond rdtype A ipv4
if dns_answer.rrset.ttl != DNS_SERVER_REDIRECT_TTL:
raise Exception("Error: The dns_action_deny_subaction_redirect_a check failure: ttl(%s) is not DNS_SERVER_REDIRECT_TTL(%d)"%(dns_answer.rrset.ttl, DNS_SERVER_REDIRECT_TTL))
return
for i in dns_answer.response.answer:
for j in i.items:
if j.rdtype == 1: #'rdtype is A: ipv4'
if j.address == "99.99.99.99":
raise Exception(DnsARequestFireWallDenyRedirect)
else:
raise Exception("Error: The dns request rdtype A drop redirect check failure: respond value error")
else:
raise Exception("Error: The dns request rdtype A drop redirect check failure: respond rdtype error")
def dns_action_deny_subaction_redirect_aaaa(self):
dns_resolver=dns.resolver.Resolver()
dns_resolver.nameservers = DNS_SERVER_IP
dns_resolver.timeout = float(3)
dns_resolver.lifetime = float(3)
try:
dns_answer = dns_resolver.query("www.2test-ipv6.com", 'AAAA')
except dns.exception.DNSException as errorinfo:
raise Exception("Error: The dns_action_deny_subaction_redirect_aaaa check failure, code: %s" % errorinfo)
else: # drop-redirect and respond rdtype A ipv6
if dns_answer.rrset.ttl != DNS_SERVER_REDIRECT_TTL:
raise Exception("Error: The dns_action_deny_subaction_redirect_aaaa check failure: ttl(%s) is not DNS_SERVER_REDIRECT_TTL(%d)"%(dns_answer.rrset.ttl, DNS_SERVER_REDIRECT_TTL))
return
for i in dns_answer.response.answer:
for j in i.items:
if j.rdtype == 28: #'rdtype is A: ipv6'
if j.address == "99:99::99:99":
raise Exception(DnsAAAARequestFireWallDenyRedirect)
else:
raise Exception("Error: The dns request rdtype AAAA drop redirect check failure: respond value error")
else:
raise Exception("Error: The dns request rdtype AAAA drop redirect check failure: respond rdtype error")
def dns_action_deny_subaction_redirect_a_rang_ttl(self):
dns_resolver=dns.resolver.Resolver()
dns_resolver.nameservers = DNS_SERVER_IP
dns_resolver.timeout = float(3)
dns_resolver.lifetime = float(3)
try:
dns_answer = dns_resolver.query("www.4test-ipv4.com", 'A')
except dns.exception.DNSException as errorinfo:
raise Exception("Error: The dns_action_deny_subaction_redirect_a_rang_ttl check failure, code: %s" % errorinfo)
else: # drop-redirect and respond rdtype A ipv4
if DNS_SERVER_REDIRECT_RANGE_LOW > dns_answer.rrset.ttl or dns_answer.rrset.ttl > DNS_SERVER_REDIRECT_RANGE_HIGH:
raise Exception("Error: The dns_action_deny_subaction_redirect_a_rang_ttl check failure: ttl(%d) is not DNS_SERVER_REDIRECT_RANG_TTL(%d-%d)"%(dns_answer.rrset.ttl,DNS_SERVER_REDIRECT_RANGE_LOW, DNS_SERVER_REDIRECT_RANGE_HIGH))
return
for i in dns_answer.response.answer:
for j in i.items:
if j.rdtype == 1: #'rdtype is A: ipv4'
if j.address == "99.99.99.99":
raise Exception(DnsARequestFireWallDenyRedirectRangTTL)
else:
raise Exception("Error: The dns request rdtype A drop redirect range ttl check failure: respond value error")
else:
raise Exception("Error: The dns request rdtype A drop redirect range ttl check failure: respond rdtype error")
def dns_action_deny_subaction_redirect_aaaa_rang_ttl(self):
dns_resolver=dns.resolver.Resolver()
dns_resolver.nameservers = DNS_SERVER_IP
dns_resolver.timeout = float(3)
dns_resolver.lifetime = float(3)
try:
dns_answer = dns_resolver.query("www.4test-ipv6.com", 'AAAA')
except dns.exception.DNSException as errorinfo:
raise Exception("Error: The dns_action_deny_subaction_redirect_aaaa range ttl check failure, code: %s" % errorinfo)
else: # drop-redirect and respond rdtype A ipv6
if DNS_SERVER_REDIRECT_RANGE_LOW > dns_answer.rrset.ttl or dns_answer.rrset.ttl > DNS_SERVER_REDIRECT_RANGE_HIGH:
raise Exception("Error: The dns_action_deny_subaction_redirect_aaaa_rang_ttl check failure: ttl(%d) is not DNS_SERVER_REDIRECT_RANG_TTL(%d-%d)"%(dns_answer.rrset.ttl, DNS_SERVER_REDIRECT_RANGE_LOW, DNS_SERVER_REDIRECT_RANGE_HIGH))
return
for i in dns_answer.response.answer:
for j in i.items:
if j.rdtype == 28: #'rdtype is A: ipv6'
if j.address == "99:99::99:99":
raise Exception(DnsAAAARequestFireWallDenyRedirectRangTTL)
else:
raise Exception("Error: The dns request rdtype AAAA drop redirect check failure: respond value error")
else:
raise Exception("Error: The dns request rdtype AAAA drop redirect check failure: respond rdtype error")
def dns_action_allow_rdtype_a(self):
dns_resolver=dns.resolver.Resolver()
dns_resolver.nameservers = DNS_SERVER_IP
dns_resolver.timeout = float(3)
dns_resolver.lifetime = float(3)
try:
dns_answer = dns_resolver.query("www.1test-ipv4.com", 'A')
except dns.exception.DNSException as errorinfo:
raise Exception("Error: The dns request rdtype A allow check failure, code: %s" % errorinfo)
else:
if dns_answer.rrset.ttl != DNS_SERVER_ALLOW_TTL:
raise Exception("Error: The dns request rdtype A allow check failure: ttl(%d) is not DNS_SERVER_ALLOW_TTL(%d)"%(dns_answer.rrset.ttl, DNS_SERVER_ALLOW_TTL))
return
for i in dns_answer.response.answer:
for j in i.items:
if j.rdtype == 1: #'rdtype is A: ipv4'
if j.address == "10.1.2.3":
raise Exception(DnsARequestFirewallAllow)
else:
raise Exception("Error: The dns request rdtype A allow check failure: respond value error")
else:
raise Exception("Error: The dns request rdtype A allow check failure: respond rdtype error")
def dns_action_allow_rdtype_aaaa(self):
dns_resolver=dns.resolver.Resolver()
dns_resolver.nameservers = DNS_SERVER_IP
dns_resolver.timeout = float(3)
dns_resolver.lifetime = float(3)
try:
dns_answer = dns_resolver.query("www.1test-ipv6.com", 'AAAA')
except dns.exception.DNSException as errorinfo:
raise Exception("Error: The dns request rdtype AAAA allow check failure, code: %s" % errorinfo)
else:
if dns_answer.rrset.ttl != DNS_SERVER_ALLOW_TTL:
raise Exception("Error: The dns request rdtype A allow check failure: ttl(%d) is not DNS_SERVER_ALLOW_TTL(%d)"%(dns_answer.rrset.ttl, DNS_SERVER_ALLOW_TTL))
return
for i in dns_answer.response.answer:
for j in i.items:
if j.rdtype == 28: #'rdtype is AAAA: ipv6'
if j.address == "11aa:11:22::33":
raise Exception(DnsAAAARequestFirewallAllow)
else:
raise Exception("Error: The dns request rdtype AAAA allow check failure: respond value error")
else:
raise Exception("Error: The dns request rdtype AAAA allow check failure: response rdtype error")
def dns_action_allow_rdtype_cname(self):
dns_resolver=dns.resolver.Resolver()
dns_resolver.nameservers = DNS_SERVER_IP
dns_resolver.timeout = float(3)
dns_resolver.lifetime = float(3)
try:
dns_answer = dns_resolver.query("www.1test-cname.com", 'CNAME')
except dns.exception.DNSException as errorinfo:
raise Exception("Error: The dns request rdtype CNAME allow check failure, code: %s" % errorinfo)
else:
if dns_answer.rrset.ttl != DNS_SERVER_ALLOW_TTL:
raise Exception("Error: The dns request rdtype A allow check failure: ttl(%d) is not DNS_SERVER_ALLOW_TTL(%d)"%(dns_answer.rrset.ttl, DNS_SERVER_ALLOW_TTL))
return
for i in dns_answer.response.answer:
for j in i.items:
if j.rdtype == 5: #'CNAME: tag(www.xxx.com)'
m=str(j)
if m == "www.1testanswer-cname.com.":
raise Exception(DnsCNAMERequestFirewallAllow)
else:
raise Exception("Error: The dns request rdtype CNAME allow check failure: respond value error")
else:
raise Exception("Error: The dns request rdtype CNAME allow check failure: respond rdtype error")
class SSLCheckRequestBuild:
@@ -548,6 +768,46 @@ class SslFirewallActionBuild:
class SslUnitTest(unittest.TestCase):
def test_dnsRequest_deny_drop(self):
dnsHandler = DNSCheckRequestBuild()
with self.assertRaisesRegex(Exception, DnsRequestFirewallDenyDrop):
dnsHandler.dns_action_deny_subaction_drop()
def test_dnsRequest_deny_redirect_a(self):
dnsHandler = DNSCheckRequestBuild()
with self.assertRaisesRegex(Exception, DnsARequestFireWallDenyRedirect):
dnsHandler.dns_action_deny_subaction_redirect_a()
def test_dnsRequest_deny_redirect_aaaa(self):
dnsHandler = DNSCheckRequestBuild()
with self.assertRaisesRegex(Exception, DnsAAAARequestFireWallDenyRedirect):
dnsHandler.dns_action_deny_subaction_redirect_aaaa()
def test_dnsRequest_deny_redirect_a_range_ttl(self):
dnsHandler = DNSCheckRequestBuild()
with self.assertRaisesRegex(Exception, DnsARequestFireWallDenyRedirectRangTTL):
dnsHandler.dns_action_deny_subaction_redirect_a_rang_ttl()
def test_dnsRequest_deny_redirect_aaaa_range_ttl(self):
dnsHandler = DNSCheckRequestBuild()
with self.assertRaisesRegex(Exception, DnsAAAARequestFireWallDenyRedirectRangTTL):
dnsHandler.dns_action_deny_subaction_redirect_aaaa_rang_ttl()
def test_dnsRequest_allow_rdtype_a(self):
dnsHandler = DNSCheckRequestBuild()
with self.assertRaisesRegex(Exception, DnsARequestFirewallAllow):
dnsHandler.dns_action_allow_rdtype_a()
def test_dnsRequest_allow_rdtype_aaaa(self):
dnsHandler = DNSCheckRequestBuild()
with self.assertRaisesRegex(Exception, DnsAAAARequestFirewallAllow):
dnsHandler.dns_action_allow_rdtype_aaaa()
def test_dnsRequest_allow_rdtype_cname(self):
dnsHandler = DNSCheckRequestBuild()
with self.assertRaisesRegex(Exception, DnsCNAMERequestFirewallAllow):
dnsHandler.dns_action_allow_rdtype_cname()
def test_securityPolicy_bypass(self):
sslHandler = SSLCheckRequestBuild()
with self.assertRaisesRegex(Exception, ssl_bypass_info_re):
@@ -764,6 +1024,14 @@ class TsgDiagnoseRun:
def _init_suite(self):
self.suite = unittest.TestSuite()
self.suite._cleanup = False
self._add_suite('test_dnsRequest_deny_drop')
self._add_suite('test_dnsRequest_deny_redirect_a')
self._add_suite('test_dnsRequest_deny_redirect_aaaa')
self._add_suite('test_dnsRequest_deny_redirect_a_range_ttl')
self._add_suite('test_dnsRequest_deny_redirect_aaaa_range_ttl')
self._add_suite('test_dnsRequest_allow_rdtype_a')
self._add_suite('test_dnsRequest_allow_rdtype_aaaa')
self._add_suite('test_dnsRequest_allow_rdtype_cname')
self._add_suite('test_securityPolicy_bypass')
self._add_suite('test_securityPolicy_intercept')
self._add_suite('test_securityPolicy_intercept_certerrExpired')
@@ -887,4 +1155,4 @@ class TsgDiagnoseRun:
if __name__ == '__main__':
tsg_diagnose_run = TsgDiagnoseRun()
tsg_diagnose_run.execute_suite_tsg_diagnose()
tsg_diagnose_run.execute_suite_tsg_diagnose()