David/IoT_PreTester_V46_Demo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

analyze performance upgrade

Browse Source
This commit is contained in:
DaStra3003
2026-09-23 08:01:49 +02:00
parent aabe35fead
commit 7f7b753350
51 changed files with 5827 additions and 1372 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_project/analyzer/tests/test_analyzer.py
+15 -13
View File
@@ -48,7 +48,7 @@ class PlaintextCredentialTests(unittest.TestCase):
authorization=f"Basic {token}", src_ip="10.0.0.5", dst_ip="93.184.216.34",
dst_port=80))
res = run(ctx)
f = next(f for f in res.findings if "HTTP Basic" in f.title)
f = next(f for f in res.findings if "Basic authentication over HTTP" in f.title)
self.assertEqual(f.evidence["username"], "bob")
@@ -76,7 +76,8 @@ class InsecureProtocolTests(unittest.TestCase):
ctx = CaptureContext()
ctx.conversations.append(Conversation(
src_ip="192.168.0.2", dst_ip="192.168.0.3", src_port=50000, dst_port=23,
protocol="tcp", packets=10, bytes=500))
protocol="tcp", packets=10, bytes=500, payload_bytes=200, synack_count=1,
app_protocols={"telnet"}))
res = run(ctx)
f = next(f for f in res.findings if f.category == Category.INSECURE_PROTOCOL)
self.assertEqual(f.severity, Severity.CRITICAL)
@@ -88,7 +89,8 @@ class TlsWeaknessTests(unittest.TestCase):
ctx = CaptureContext()
ctx.tls_sessions.append(TlsSession(
src_ip="10.0.0.1", dst_ip="1.2.3.4", dst_port=443,
version="TLS 1.0", cipher="TLS_RSA_WITH_RC4_128_SHA", sni="legacy.example"))
version="TLS 1.0", cipher="TLS_RSA_WITH_RC4_128_SHA", sni="legacy.example",
server_hello_seen=True))
res = run(ctx)
titles = [f.title for f in res.findings if f.category == Category.TLS_WEAKNESS]
self.assertTrue(any("Obsolete TLS" in t for t in titles))
@@ -337,7 +339,7 @@ class TlsCertTests(unittest.TestCase):
src_ip="10.0.0.1", dst_ip="1.2.3.4", dst_port=443,
version="TLS 1.2", sni="iot.local",
cert_subject_cn="iot.local", cert_issuer_cn="iot.local",
cert_self_signed=True))
cert_self_signed=True, cert_seen=True))
res = run(ctx)
self.assertTrue(any("self-signed" in f.title for f in res.findings))
@@ -345,20 +347,20 @@ class TlsCertTests(unittest.TestCase):
ctx = CaptureContext()
ctx.tls_sessions.append(TlsSession(
src_ip="10.0.0.1", dst_ip="1.2.3.4", dst_port=443,
version="TLS 1.2", cert_expired=True))
version="TLS 1.2", cert_expired=True, cert_seen=True))
res = run(ctx)
self.assertTrue(any("expired" in f.title for f in res.findings))
class CertHelperTests(unittest.TestCase):
def test_self_signed_helper(self):
from analyzer.extract_pyshark import _cert_is_self_signed
from analyzer.extract_tshark import _cert_is_self_signed
self.assertTrue(_cert_is_self_signed("device.local", "device.local"))
self.assertFalse(_cert_is_self_signed("device.local", "DigiCert CA"))
self.assertIsNone(_cert_is_self_signed("device.local", None))
def test_expired_helper(self):
from analyzer.extract_pyshark import _cert_is_expired
from analyzer.extract_tshark import _cert_is_expired
# The certificate expired before the 2025 capture timestamp.
now = 1735689600.0 # 2025-01-01 UTC
self.assertTrue(_cert_is_expired("2020-01-01 00:00:00", now))
@@ -366,7 +368,7 @@ class CertHelperTests(unittest.TestCase):
self.assertIsNone(_cert_is_expired(None, now))
def test_parse_cert_time_formats(self):
from analyzer.extract_pyshark import _parse_cert_time
from analyzer.extract_tshark import _parse_cert_time
self.assertIsNotNone(_parse_cert_time("2020-01-01 00:00:00"))
self.assertIsNotNone(_parse_cert_time("Jan 1 00:00:00 2020 GMT"))
self.assertIsNotNone(_parse_cert_time("200101000000Z"))
@@ -490,7 +492,7 @@ class UserAgentTests(unittest.TestCase):
class DnsQtypeTests(unittest.TestCase):
def test_qtype_name_mapping(self):
from analyzer.extract_pyshark import _dns_qtype_name
from analyzer.extract_tshark import _dns_qtype_name
self.assertEqual(_dns_qtype_name("16"), "TXT")
self.assertEqual(_dns_qtype_name("1"), "A")
self.assertEqual(_dns_qtype_name("txt"), "TXT")
@@ -516,14 +518,14 @@ class DnsQtypeTests(unittest.TestCase):
class DownloadHelperTests(unittest.TestCase):
def test_is_download_content_type(self):
from analyzer.extract_pyshark import _is_download_content_type
from analyzer.extract_tshark import _is_download_content_type
self.assertTrue(_is_download_content_type("application/octet-stream"))
self.assertTrue(_is_download_content_type("application/x-msdownload"))
self.assertFalse(_is_download_content_type("text/html; charset=utf-8"))
self.assertFalse(_is_download_content_type(None))
def test_filename_from_disposition(self):
from analyzer.extract_pyshark import _filename_from_disposition
from analyzer.extract_tshark import _filename_from_disposition
self.assertEqual(_filename_from_disposition('attachment; filename="update.exe"'), "update.exe")
self.assertEqual(_filename_from_disposition("attachment; filename=payload.bin"), "payload.bin")
self.assertIsNone(_filename_from_disposition("inline"))
@@ -537,7 +539,7 @@ class FileTransferDownloadTests(unittest.TestCase):
filename=None, extension=None, content_type="application/x-msdownload",
protocol="http", src_ip="1.2.3.4", dst_ip="10.0.0.5", direction="download"))
res = run(ctx)
self.assertTrue(any("harmful file" in f.title for f in res.findings))
self.assertTrue(any("Executable" in f.title for f in res.findings))
def test_disposition_filename_extension_flagged(self):
from analyzer.context import FileTransfer
@@ -765,7 +767,7 @@ class DemoPayloadRegressionTests(unittest.TestCase):
self.assertTrue(any("40.101.70.178" in u for u in urls))
def test_hexish_payload_decoder(self):
from analyzer.extract_pyshark import _bytes_from_hexish
from analyzer.extract_tshark import _bytes_from_hexish
# pyshark commonly exposes payload bytes as colon-separated hex.
self.assertEqual(_bytes_from_hexish("47:45:54"), "GET")
self.assertEqual(_bytes_from_hexish("already text"), "already text")