Merge pull request #20 from MetLife/develop

version 2.0

.gitignore

@@ -90,4 +90,7 @@ test-output.xml
 # vim swap
 *.swp
 
+# pylint
+*.pylintrc
+
 todo.txt

CHANGELOG.md

@@ -10,4 +10,9 @@
 - Added some type annotations and more consistent error for the main function
 
 # v1.3.0
-- Added the ability to scan by IP
+- Added the ability to scan by IP
+
+# v2.0.0
+- Upgraded [SSLyze](https://github.com/nabla-c0d3/sslyze) to 3.x
+- Added several TLS 1.2 ciphers to the "policy" scan type as "weak"
+- Added scan type and port to result set

README.md

@@ -6,9 +6,11 @@ SSLChecker is a serverless API written in Python and running on Azure Functions.
 
 ## Pre-requisites
 
-Development - To set up a local development environment, follow the guidance from Microsoft [here](https://docs.microsoft.com/en-us/azure/azure-functions/functions-create-first-azure-function-azure-cli?pivots=programming-language-python&tabs=bash%2Cbrowser).
+__Python__ - Python 3.7 (64-bit) and above.
 
-Deployment - As part of the above setup, you will be able to deploy to Azure using the azure-cli. Additionally, Azure DevOps or another CI/CD tool is capable of deploying to Azure.
+__Development__ - To set up a local development environment, follow the guidance from Microsoft [here](https://docs.microsoft.com/en-us/azure/azure-functions/functions-create-first-azure-function-azure-cli?pivots=programming-language-python&tabs=bash%2Cbrowser).
+
+__Deployment__ - As part of the above setup, you will be able to deploy to Azure using the azure-cli. Additionally, Azure DevOps or another CI/CD tool is capable of deploying to Azure.
 
 ## Usage
 
@@ -18,7 +20,7 @@ Invoke the function on the command line using curl:
 
 There are four parts to pass to the URI: scan, view, target, and port.
 
-"scan" is the type of scan: policy or full. Currently, the default policy prohibits using SSL 2.0/3.0 and TLS 1.0/1.1, so the policy scan will identify which unsupported ciphers are in use, if any. A full scan will report back all supported ciphers. In a future release I will make this configurable.
+"scan" is the type of scan: policy or full. Currently, the default policy prohibits using SSL 2.0/3.0, TLS 1.0/1.1 and some TLS 1.2 ciphers, so the policy scan will identify which unsupported ciphers are in use, if any. A full scan will report back all supported ciphers.
 
 Since corporations often use [split-view DNS](https://en.wikipedia.org/wiki/Split-horizon_DNS), "view" in this context is the network viewpoint you want to scan, either internal or external. This is accomplished by specifying a valid DNS server to use for name resolution. The default value for external will use OpenDNS (e.g. 208.67.222.222). The default for internal will be 0.0.0.0 and will result in an error if a scan is attempted and no internal DNS server is specified. Please modify the config.ini file to use an internal DNS server.
 

SSLChecker/requirements.txt

@@ -1,13 +1,7 @@
 azure-functions
-asn1crypto==1.2.0
-cffi==1.13.2
-cryptography==2.5
 dnspython==1.16.0
-nassl==2.2.0
-pycparser==2.19
 requests==2.22.0
-sslyze==2.1.4
-tls-parser==1.2.1
-validators==0.14.1
+sslyze==3.06
+validators==0.15.0
 typing-extensions==3.7.4.2
-pytest==5.4.1
+pytest==5.4.3

SSLChecker/sharedcode/results.py

@@ -5,8 +5,8 @@ def set_error(error_type, message):
     return {"Error Type": error_type, "Message": message}
 
 
-def new():
-    return {"Target": None, "IP": None, "MD5": None, "View": None, "Results": []}
+def new_result_set():
+    return {"Target": None, "IP": None, "MD5": None, "Scan": None, "View": None, "Results": []}
 
 
 def set_result(results, key, value):

SSLChecker/sharedcode/scanner.py

@@ -1,45 +1,51 @@
 from hashlib import md5
 
-from sslyze.server_connectivity_tester import (
+from sslyze import (
+    ServerNetworkLocationViaDirectConnection,
     ServerConnectivityTester,
-    ServerConnectivityError,
-    ConnectionToServerTimedOut,
+    errors,
+    ScanCommand,
+    Scanner,
+    ServerScanRequest,
 )
-from sslyze.ssl_settings import TlsWrappedProtocolEnum
-from sslyze.plugins.openssl_cipher_suites_plugin import (
-    Sslv20ScanCommand,
-    Sslv30ScanCommand,
-    Tlsv10ScanCommand,
-    Tlsv11ScanCommand,
-    Tlsv12ScanCommand,
-    Tlsv13ScanCommand,
-)
-from sslyze.synchronous_scanner import SynchronousScanner
 
 from . import results
 from .errors import ConnectionError
 
-# Policy prohibits the use of SSL 2.0/3.0 and TLS 1.0
-ciphersuites = {
+# Policy prohibits the use of SSL 2.0/3.0, TLS 1.0/1.1 and
+# some TLS 1.2 cipher suites
+CIPHER_SUITES = {
     "policy": [
-        Sslv20ScanCommand(),
-        Sslv30ScanCommand(),
-        Tlsv10ScanCommand(),
-        Tlsv11ScanCommand(),
+        ScanCommand.SSL_2_0_CIPHER_SUITES,
+        ScanCommand.SSL_3_0_CIPHER_SUITES,
+        ScanCommand.TLS_1_0_CIPHER_SUITES,
+        ScanCommand.TLS_1_1_CIPHER_SUITES,
+        ScanCommand.TLS_1_2_CIPHER_SUITES,
     ],
     "full": [
-        Sslv20ScanCommand(),
-        Sslv30ScanCommand(),
-        Tlsv10ScanCommand(),
-        Tlsv11ScanCommand(),
-        Tlsv12ScanCommand(),
-        Tlsv13ScanCommand(),
+        ScanCommand.SSL_2_0_CIPHER_SUITES,
+        ScanCommand.SSL_3_0_CIPHER_SUITES,
+        ScanCommand.TLS_1_0_CIPHER_SUITES,
+        ScanCommand.TLS_1_1_CIPHER_SUITES,
+        ScanCommand.TLS_1_2_CIPHER_SUITES,
+        ScanCommand.TLS_1_3_CIPHER_SUITES,
     ],
 }
 
-# sslyze config
-SynchronousScanner.DEFAULT_NETWORK_RETRIES = 1
-SynchronousScanner.DEFAULT_NETWORK_TIMEOUT = 3
+# Currently, only The following TLS 1.2 ciphers are considered "strong"
+ALLOWED_TLS12_CIPHERS = {
+    "TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256",
+    "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256",
+    "TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384",
+    "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384",
+    "TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256",
+    "TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256",
+    "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256",
+    "TLS_DHE_RSA_WITH_AES_128_GCM_SHA256",
+    "TLS_DHE_RSA_WITH_AES_256_GCM_SHA384",
+    "TLS_DHE_RSA_WITH_CHACHA20_POLY1305_SHA256",
+}
+
 
 ERROR_MSG_CONNECTION_TIMEOUT = "TCP connection to {}:{} timed-out".format
 ERROR_MSG_UNKNOWN_CONNECTION = (
@@ -51,44 +57,62 @@ def scan(target, ip, port, view, suite):
     """ Five inputs: web site name, ip, port
     split-dns view, and cipher suite """
 
+    server_location = ServerNetworkLocationViaDirectConnection(target, port, ip)
+
+    # This line checks to see if the host is online
     try:
-        server_tester = ServerConnectivityTester(
-            hostname=target,
-            ip_address=ip,
-            port=port,
-            tls_wrapped_protocol=TlsWrappedProtocolEnum.HTTPS,
-        )
-        # This line checks to see if the host is online
-        server_info = server_tester.perform()
-        ip = server_info.ip_address
-    # Could not establish an SSL connection to the server
-    except ConnectionToServerTimedOut:
+        server_info = ServerConnectivityTester().perform(server_location)
+    except errors.ConnectionToServerTimedOut:
         raise ConnectionError(
             "Connection Timeout", ERROR_MSG_CONNECTION_TIMEOUT(target, port)
         )
-    except ServerConnectivityError:
+    except errors.ConnectionToServerFailed:
         raise ConnectionError(
-            "Unknow Connection Error", ERROR_MSG_UNKNOWN_CONNECTION(target, port)
+            "Unknown Connection Error", ERROR_MSG_UNKNOWN_CONNECTION(target, port)
         )
 
     # Create a new results dictionary
-    scan_output = results.new()
+    scan_output = results.new_result_set()
 
     # I hash the combination of hostname and ip for tracking
     key = md5((target + ip).encode("utf-8")).hexdigest()
     results.set_result(scan_output, "MD5", key)
-    results.set_result(scan_output, "Target", f"{target}")
-    results.set_result(scan_output, "IP", ip)
+    results.set_result(scan_output, "Target", f"{target}:{port}")
+    results.set_result(scan_output, "IP", f"{ip}:{port}")
+    results.set_result(scan_output, "Scan", suite)
     results.set_result(scan_output, "View", view)
 
-    for suite in ciphersuites.get(suite):
-        synchronous_scanner = SynchronousScanner()
-        scan_result = synchronous_scanner.run_scan_command(server_info, suite)
+    scanner = Scanner()
+    server_scan_req = ServerScanRequest(
+        server_info=server_info, scan_commands=CIPHER_SUITES.get(suite)
+    )
+    scanner.queue_scan(server_scan_req)
+
+    for result in scanner.get_results():
+        for cipher_suite in CIPHER_SUITES.get(suite):
+            scan_result = result.scan_commands_results[cipher_suite]
 
-        for cipher in scan_result.accepted_cipher_list:
-            results.set_ciphers(
-                scan_output, {"Version": cipher.ssl_version.name, "Cipher": cipher.name}
-            )
+            for accepted_cipher_suite in scan_result.accepted_cipher_suites:
+                if suite == "policy" and scan_result.tls_version_used.name == "TLS_1_2":
+                    if (
+                        accepted_cipher_suite.cipher_suite.name
+                        not in ALLOWED_TLS12_CIPHERS
+                    ):
+                        results.set_ciphers(
+                            scan_output,
+                            {
+                                "Version": f"{scan_result.tls_version_used.name}",
+                                "Cipher": f"{accepted_cipher_suite.cipher_suite.name}",
+                            },
+                        )
+                else:
+                    results.set_ciphers(
+                        scan_output,
+                        {
+                            "Version": f"{scan_result.tls_version_used.name}",
+                            "Cipher": f"{accepted_cipher_suite.cipher_suite.name}",
+                        },
+                    )
 
     if len(scan_output["Results"]) == 0:
         results.set_result(scan_output, "Results", "No Policy Violations")

tests/pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+filterwarnings = 
+    ignore:PY_SSIZE_T_CLEAN:DeprecationWarning

tests/test_SSLChecker.py

@@ -15,7 +15,7 @@ def test_policy_external_no_violations():
             url='/api/',
             route_params={'scan': 'policy',
                           'view': 'external',
-                          'target': 'microsoft.com'}
+                          'target': 'api.metlife.com'}
             )
 
     # Call the function
@@ -383,7 +383,7 @@ def test_policy_external_by_ip_no_violations():
             url='/api/',
             route_params={'scan': 'policy',
                           'view': 'external',
-                          'target': '140.82.113.4'}
+                          'target': '216.163.251.205'}
         )
     # Call the function.
     resp = main(req)