Input validation testing on key endpoints

documents/README_SecurityAudit.md

@@ -0,0 +1,7 @@
+This summarises issues referred to in [Security Audit of Place Places Server and Application](Security Audit of Place Places Server and Application.pdf),
+
+* Sidejacking risk is covered in this jira: https://pathcheck.atlassian.net/browse/PLACES-324 and https://pathcheck.atlassian.net/browse/PLACES-298
+* Local storage here: https://pathcheck.atlassian.net/browse/PLACES-346
+* Content security policy: https://pathcheck.atlassian.net/browse/PLACES-272 and https://pathcheck.atlassian.net/browse/PLACES-339
+* Mitigate cross-site request forgery (CSRF) attacks: https://pathcheck.atlassian.net/browse/PLACES-340
+* Rate limit the login endpoint:  https://pathcheck.atlassian.net/browse/PLACES-337

dynamic_testing/JSONFuzzing/CovidApp.py

@@ -0,0 +1,158 @@
+#!/usr/bin/python3
+import http.client
+import json
+import time
+
+class CovidApp():
+    token = ""
+    spl_url = "zeus.safeplaces.extremesolution.com"
+    pe_url= "hermes.safeplaces.extremesolution.com"
+    code = 0;
+
+
+    def validate_results(self, response):
+        print("duration" + str(response['duration']))
+        assert response['duration'] < 5, "Over time.  HTTP status: " + str(response['status'])
+        print("status" + str(response['status']))
+        assert response['status'] in range (200, 299), "Bad HTTP status: " + str(response['status'])
+        return response
+
+    def validate_results_expect_error(self, response):
+        print("duration" + str(response['duration']))
+        assert response['duration'] < 70
+        print("status" + str(response['status']))
+        assert response['status'] in (401, 504, 500, 501, 400), "Bad HTTP status: " + str(response['status'])
+        print("OK response: " + str(response['status']))
+        return response
+
+    def authenticated_post(self, endpoint, payload):
+        conn = http.client.HTTPSConnection(self.spl_url)
+        headers = {
+            'content-type': "application/json",
+            'cache-control': "no-cache",
+            'Authorization': "Bearer " + self.token,
+        }
+
+        start_time = time.time()
+        print("Calling " + endpoint + " payload:" + str(payload) + " headers: " + str(headers))
+        conn.request("POST", endpoint, payload, headers)
+        result = {}
+        result['response'] = conn.getresponse()
+        result['status'] = result['response'].status
+        result['duration'] = time.time() - start_time
+        return self.validate_results(result)
+
+    def authenticated_post_with_payload(self, endpoint, payload):
+        conn = http.client.HTTPSConnection(self.spl_url)
+        headers = {
+            'content-type': "application/json",
+            'cache-control': "no-cache",
+            'Authorization': "Bearer " + self.token,
+        }
+
+        start_time = time.time()
+        print("Calling " + endpoint + " payload:" + str(payload) + " headers: " + str(headers))
+        conn.request("POST", endpoint, payload, headers)
+        result = {}
+        result['response'] = conn.getresponse()
+        result['status'] = result['response'].status
+        result['duration'] = time.time() - start_time
+        return self.validate_results(result)
+
+    def public_unauthenticated_post(self, endpoint, payload):
+        conn = http.client.HTTPSConnection(self.pe_url)
+        headers = {
+            'content-type': "application/json",
+            'cache-control': "no-cache"
+        }
+        start_time = time.time()
+        print("Calling " + endpoint + " payload:" + str(payload) + " headers: " + str(headers))
+        conn.request("POST", endpoint, payload, headers)
+        result = {}
+        result['response'] = conn.getresponse()
+        result['status'] = result['response'].status
+        result['duration'] = time.time() - start_time
+        return self.validate_results(result)
+
+    def public_unauthenticated_post_any_result(self, endpoint, payload):
+        conn = http.client.HTTPSConnection(self.pe_url)
+        headers = {
+            'content-type': "application/json",
+            'cache-control': "no-cache"
+        }
+        start_time = time.time()
+        print("Calling " + endpoint + " payload:" + str(payload) + " headers: " + str(headers))
+        conn.request("POST", endpoint, payload, headers)
+        result = {}
+        result['response'] = conn.getresponse()
+        result['status'] = result['response'].status
+        result['duration'] = time.time() - start_time
+        return self.validate_results(result)
+
+    def spl_login_post(self, endpoint, payload):
+        conn = http.client.HTTPSConnection(self.spl_url)
+        headers = {
+            'content-type': "application/json",
+            'cache-control': "no-cache"
+        }
+        start_time = time.time()
+        print("Calling " + endpoint + " payload:" + str(payload) + " headers: " + str(headers))
+        conn.request("POST", endpoint, payload, headers)
+        result = {}
+        result['response'] = conn.getresponse()
+        result['status'] = result['response'].status
+        result['duration'] = time.time() - start_time
+        return self.validate_results(result)
+
+    def spl_login_post_expect_error(self, endpoint, payload):
+        print("Calling: " + str(self.spl_url) + str(endpoint))
+        conn = http.client.HTTPSConnection(self.spl_url)
+        headers = {
+            'content-type': "application/json",
+            'cache-control': "no-cache"
+        }
+        start_time = time.time()
+        conn.request("POST", endpoint, payload, headers)
+        result = {}
+        result['response'] = conn.getresponse()
+        result['status'] = result['response'].status
+        result['duration'] = time.time() - start_time
+        return self.validate_results_expect_error(result)
+
+    def login_as_contact_tracer(self, user, pw):
+        payload = json.loads("{\"username\": \"" + user + "\", \"password\":\"" + pw + "\"}")
+        result = self.spl_login_post("/login", json.dumps(payload))
+        self.token = json.loads(result['response'].read())['token']
+
+    def get_an_access_code(self):
+        result = self.authenticated_post("/access-code", "")
+        self.code = json.loads(result['response'].read())['accessCode']
+        return self.code
+
+    def get_access_code_with_payload(self, payload):
+        result = self.authenticated_post_with_payload("/access-code", payload)
+        self.code = json.loads(result['response'].read())['accessCode']
+
+    def user_consent(self, code):
+        payload = {}
+        payload['accessCode'] = self.code
+        payload['consent'] = True
+        self.public_unauthenticated_post("/consent", json.dumps(payload) )
+
+    def upload_data(self, code, json_data):
+        payload = {}
+        payload['accessCode'] = self.code
+        payload['concernPoints'] = json_data
+        result = self.public_unauthenticated_post("/upload", json.dumps(payload))
+        response = result['response'].read()
+        print("Response: " + response)
+        return response
+
+    def upload_data_any_result(self, code, json_data):
+        payload = {}
+        payload['accessCode'] = self.code
+        payload['concernPoints'] = json_data
+        result = self.public_unauthenticated_post_any_result("/upload", json.dumps(payload))
+        response = result['response'].read()
+        print("Response: " + response)
+        return response

dynamic_testing/JSONFuzzing/JSONFuzzing.md

@@ -0,0 +1,25 @@
+# OWASP principles in scope
+Verify that all input (HTML form fields, REST requests, URL parameters, HTTP headers, cookies, batch files, RSS feeds, etc) is validated using positive validation (whitelisting).
+# Endpoints in scope
+
+* Safeplaces facing /login API endpoint, and endpoint to get a new access-code
+* Public endpoint API for uploading location data
+
+# Purpose of test
+This test was to provide some assurance that these key endpoints are not vulnerable to JSON injection.  
+
+The end-points were selected as they are accessible to someone reverse engineering the mobile application, and a contact tracer working inside a healthcare authority, and represent the only data entry point to the Safe Places system in the current release.  
+
+The existence of a vulnerabilty may mean that bad data can be injected, existing data can be compromised, or the system can be otherwise subverted at the application code level.
+
+The actual payloads for the attacks were based on a list of known JSON attacks from fuzzdb.
+
+The tests require pytest, and can be run by adding "test_" to the start of the methods name in test_fuzzing.py.  For example, changing "happy_path" to "test_happy_path", then running pytest in this folder, will execute a happy path test.
+
+# Test results
+* The login endpoint was not tested with SQL injection as it relies on LDAP, and that is a test server not a recommended implementation
+* The access-code endpoint was tested to ensure that if any fuzz example was sent, an access code was still provided (as the POST payload is not used).  This test failed as 4xx errors were seen.  [See jira.}(https://pathcheck.atlassian.net/browse/PLACES-423?atlOrigin=eyJpIjoiNDNjYmIyMTEwN2Q1NDBlNjg3YWFmZTU4YmM0NjExYWUiLCJwIjoiaiJ9).  This does not mean the application is necessarily vulnerable.
+* The upload endpoint was tested to ensure that fuzz data was not propogated into the solution.  In some instances the data was accepted (2xx response), and I checked in the UI that out of all the posted fuzz examples, only one actually ended up inside the database (and accessible through the UI), and this was simply empty, rather than holding corrupt data
+
+# Further testing
+This covers a JSON payload on some priority interfaces, but not all of them.

dynamic_testing/JSONFuzzing/test_fuzzing.py

@@ -0,0 +1,81 @@
+import unittest
+import traceback
+import json
+from CovidApp import CovidApp
+
+class CovidPathTests(unittest.TestCase):
+        #def __init__(self):
+
+            #jsonPayload = ...load it...https://github.com/fuzzdb-project/fuzzdb
+
+        #Happy Path
+        #response is a tuple of status code, and time to respond
+
+        # TODO: Need a URL that returns all URLs
+        # return safeplaces-backend/oas3.yaml
+        # Write a test that locks the attack surface - is the OAS3 spec automatically generated
+
+        #Test javascript injections / nodejs applications
+
+        #Test large files - with valid data
+
+        #Test - Verify that structured data is strongly typed and validated against a defined schema including allowed characters, length and pattern
+        #(e.g. credit card numbers or telephone, or validating that two related fields are reasonable, such as checking that suburb and zip/postcode match).
+
+
+
+    def jsonfuzz_login(self):
+        app = CovidApp()
+        with open("JSON_Fuzzing.txt") as file_in:
+            lines = []
+            for payload in file_in:
+                print(payload)
+                result = app.spl_login_post_expect_error("/login", payload)
+                try:
+                    token = json.loads(result['response'].read())['token']
+                    print(token)
+                except:
+                    pass
+
+    def jsonfuzz_get_access_code(self):
+        app = CovidApp()
+        app.login_as_contact_tracer("spladmin", "password")
+        with open("JSON_Fuzzing.txt") as file_in:
+            lines = []
+            for payload in file_in:
+                print(payload)
+                result = app.get_access_code_with_payload(payload)
+                try:
+                    code = json.loads(result['response'].read())['token']
+                    print(code)
+                except:
+                    pass
+
+    def jsonfuzz_upload(self):
+        app = CovidApp()
+        app.login_as_contact_tracer("spladmin", "password")
+        with open("JSON_Fuzzing.txt") as file_in:
+            lines = []
+            for payload in file_in:
+                print(payload)
+                code = app.get_an_access_code()
+                app.user_consent(code)
+                try:
+                    result =  app.upload_data(code, payload)
+                except Exception as e:
+                    #print(result['response'].read())
+                    tb = traceback.format_exc()
+                    print(tb)
+
+
+    def happy_path(self):
+        app = CovidApp()
+        app.login_as_contact_tracer("spladmin", "password")
+        code = app.get_an_access_code()
+        app.user_consent(code)
+
+        with open('privkit31A-synthetic-REDACTED.json', 'r') as myfile:
+            data = myfile.read()
+            obj = json.loads(data)
+
+        app.upload_data(code, obj)

dynamic_testing/MSTSG_STORAGE.md

@@ -50,7 +50,7 @@ All of the OWASP principles and checks that were static, e.g code review, were c
 
 ### Dynamic Review
 
-Secure Database - I wasn't able to gain access to the secure database, but a [process]() has since been identified so this can be conducted in future checks
+Secure Database - I wasn't able to gain access to the secure database, but a [process](https://gist.github.com/troach-sf/f257bb7b80e6dddd4f3bade81b7b1410) has since been identified so this can be conducted in future checks
 
 Legacy Databases - The RKStorage, logback.db and cordova_bg_geolocation.db SQL lite databases were examined.