diff --git a/.gitignore b/.gitignore index 26fdf0fcd..82b6bbf64 100644 --- a/.gitignore +++ b/.gitignore @@ -14,6 +14,7 @@ testing/unit/ntp/output/ testing/unit/tls/output/ testing/unit/tls/tmp/ testing/unit/report/output/ +testing/unit/risk_profile/output/ *.deb make/DEBIAN/postinst diff --git a/framework/python/src/api/api.py b/framework/python/src/api/api.py index fdc3b8277..d81432321 100644 --- a/framework/python/src/api/api.py +++ b/framework/python/src/api/api.py @@ -679,7 +679,10 @@ def get_profiles_format(self, response: Response): return self.get_session().get_profiles_format() def get_profiles(self): - return self.get_session().get_profiles() + profiles = [] + for profile in self.get_session().get_profiles(): + profiles.append(json.loads(profile.to_json())) + return profiles async def update_profile(self, request: Request, response: Response): @@ -695,12 +698,6 @@ async def update_profile(self, request: Request, response: Response): return self._generate_msg(False, "Invalid request received") - # Check that profile is valid - valid_profile = self.get_session().validate_profile(req_json) - if not valid_profile: - response.status_code = status.HTTP_400_BAD_REQUEST - return self._generate_msg(False, "Invalid profile request received") - profile_name = req_json.get("name") # Check if profile exists diff --git a/framework/python/src/common/risk_profile.py b/framework/python/src/common/risk_profile.py index 021a1c5b5..4eca18b88 100644 --- a/framework/python/src/common/risk_profile.py +++ b/framework/python/src/common/risk_profile.py @@ -12,38 +12,235 @@ # See the License for the specific language governing permissions and # limitations under the License. """Stores additional information about a device's risk""" - from datetime import datetime -import os +from dateutil.relativedelta import relativedelta +from common import logger +import json -PROFILES_PATH = 'local/risk_profiles' +LOGGER = logger.get_logger('risk_profile') class RiskProfile(): """Python representation of a risk profile""" - def __init__(self, json_data): - self.name = json_data['name'] + def __init__(self, profile_json=None, profile_format=None): + + if profile_json is None or profile_format is None: + return + + self.name = profile_json['name'] + self.created = datetime.now() + self.version = profile_json['version'] + self.questions = profile_json['questions'] + self.status = None + self.risk = None + + self._validate(profile_json, profile_format) + self._update_risk(profile_format) + + # Load a profile without modifying the created date + # but still validate the profile + def load(self, profile_json, profile_format): + self.name = profile_json['name'] + self.created = datetime.strptime( + profile_json['created'], '%Y-%m-%d') + self.version = profile_json['version'] + self.questions = profile_json['questions'] + self.status = None + + self._validate(profile_json, profile_format) + self._update_risk(profile_format) + + return self - if 'status' in json_data: - self.status = json_data['status'] + def update(self, profile_json, profile_format): + + # Construct a new profile from json data + new_profile = RiskProfile(profile_json, profile_format) + + # Check if name has changed + self.name = profile_json[ + 'rename'] if 'rename' in profile_json else profile_json['name'] + + # Update this profile with newly created profile data + self.version = new_profile.version + self.created = new_profile.created + self.questions = new_profile.questions + self.status = new_profile.status + + def _validate(self, profile_json, profile_format): + if self._valid(profile_json, profile_format): + self.status = 'Expired' if self._expired() else 'Valid' else: self.status = 'Draft' - self.created = datetime.strptime(json_data['created'], - '%Y-%m-%d') - self.version = json_data['version'] - self.questions = json_data['questions'] - - def to_json(self): - json = { - 'name': self.name, - 'version': self.version, - 'created': self.created.strftime('%Y-%m-%d'), - 'status': self.status, - 'questions': self.questions - } - return json + def _update_risk(self, profile_format): + + if self.status == 'Valid': + + # Default risk = Limited + risk = 'Limited' + + # Check each question in profile + for question in self.questions: + question_text = question['question'] + + # Fetch the risk levels from the profile format + format_q = self._get_format_question( + question_text, profile_format) + + if format_q is None: + # This occurs when a question found in a current profile + # has been removed from the format (format change) + continue + + # We only want to check the select or select-multiple + # questions for now + if format_q['type'] in ['select', 'select-multiple']: + answer = question['answer'] + + question_risk = 'Limited' + + # The answer is a single string (select) + if isinstance(answer, str): + + format_option = self._get_format_question_option( + format_q, answer) + + # Format options may just be a list of strings with + # no risk attached + if format_option is None: + continue + + if 'risk' in format_option and format_option['risk'] == 'High': + question_risk = format_option['risk'] + + # A list of indexes is the answer (select-multiple) + elif isinstance(answer, list): + + format_options = format_q['options'] + + for index in answer: + option = self._get_option_from_index(format_options, index) - def get_file_path(self): - return os.path.join(PROFILES_PATH, - self.name + '.json') + if option is None: + LOGGER.error('Answer had an invalid index for question: ' + + format_q['question']) + continue + + if 'risk' in option and option['risk'] == 'High': + question_risk = 'High' + + question['risk'] = question_risk + + for question in self.questions: + if 'risk' in question and question['risk'] == 'High': + risk = 'High' + + self.risk = risk + + else: + # Remove risk + risk = None + self.risk = risk + + def _get_format_question(self, question, profile_format): + for q in profile_format: + if q['question'] == question: + return q + return None + + def _get_option_from_index(self, options, index): + i = 0 + for option in options: + if i == index: + return option + i+=1 + return None + + def _check_answer(self, question): + status = 'Limited' + if question['validation']['required']: + answer = question['answer'] + if question['type'] == 'select-multiple': + if len(answer) > 0: + status = 'High' + elif question['type'] == 'select': + if answer: + status = 'High' + return status + + def _get_profile_question(self, profile_json, question): + + for q in profile_json['questions']: + if question.lower() == q['question'].lower(): + return q + + return None + + def _get_format_question_option(self, question_obj, answer): + + for option in question_obj['options']: + + # Ignore just string lists + if isinstance(option, str): + continue + + if option['text'] == answer: + return option + + return None + + def _valid(self, profile_json, profile_format): + + # Check name field is present + if 'name' not in profile_json: + return False + + # Check questions field is present + if 'questions' not in profile_json: + return False + + all_questions_answered = True + all_questions_present = True + # Check all questions are present with answers + for format_question in profile_format: + # Check question is present + profile_question = self._get_profile_question(profile_json, + format_question['question']) + try: + required = format_question['validation']['required'] + except KeyError: + required = False + if profile_question is not None and required: + # Check answer is present + if 'answer' not in profile_question: + LOGGER.error('Missing answer for question: ' + + profile_question.get('question')) + all_questions_answered = False + elif required: + LOGGER.error('Missing question: ' + format_question.get('question')) + all_questions_present = False + + return all_questions_answered and all_questions_present + + def _expired(self): + # Calculate the date one year after the creation date + expiry_date = self.created + relativedelta(years=1) + + # Normalize the current date and time to midnight + today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) + + # Check if the current date and time is past the expiry date + return today > expiry_date + + def to_json(self, pretty=False): + json_dict = { + 'name': self.name, + 'version': self.version, + 'created': self.created.strftime('%Y-%m-%d'), + 'status': self.status, + 'risk': self.risk, + 'questions': self.questions + } + indent = 2 if pretty else None + return json.dumps(json_dict, indent=indent) diff --git a/framework/python/src/common/session.py b/framework/python/src/common/session.py index 5127cc221..038a43879 100644 --- a/framework/python/src/common/session.py +++ b/framework/python/src/common/session.py @@ -82,7 +82,11 @@ def __init__(self, root_dir): # Profiles self._profiles = [] + # Profile format that is passed to the frontend + # (excluding internal properties) self._profile_format_json = None + # Profile format used for internal validation + self._profile_format = None # System configuration self._config_file = os.path.join(root_dir, CONFIG_FILE_PATH) @@ -359,32 +363,60 @@ def _load_profiles(self): with open(os.path.join( self._root_dir, PROFILE_FORMAT_PATH ), encoding='utf-8') as profile_format_file: - self._profile_format_json = json.load(profile_format_file) + + format_json = json.load(profile_format_file) + + # Save original profile format for internal validation + self._profile_format = format_json + except (IOError, ValueError) as e: LOGGER.error( 'An error occurred whilst loading the risk assessment format') LOGGER.debug(e) + profile_format_array = [] + + # Remove internal properties + for question_obj in format_json: + new_obj = {} + for key in question_obj: + if key == 'options': + options = [] + for option in question_obj[key]: + if isinstance(option, str): + options.append(option) + else: + options.append(option['text']) + new_obj['options'] = options + else: + new_obj[key] = question_obj[key] + + profile_format_array.append(new_obj) + + self._profile_format_json = profile_format_array + # Load existing profiles LOGGER.debug('Loading risk profiles') - try: - for risk_profile_file in os.listdir(os.path.join( - self._root_dir, PROFILES_DIR - )): - LOGGER.debug(f'Discovered profile {risk_profile_file}') - - with open(os.path.join( - self._root_dir, PROFILES_DIR, risk_profile_file - ), encoding='utf-8') as f: - json_data = json.load(f) - risk_profile = RiskProfile(json_data) - risk_profile.status = self.check_profile_status(risk_profile) - self._profiles.append(risk_profile) + #try: + for risk_profile_file in os.listdir(os.path.join( + self._root_dir, PROFILES_DIR + )): + LOGGER.debug(f'Discovered profile {risk_profile_file}') - except Exception as e: - LOGGER.error('An error occurred whilst loading risk profiles') - LOGGER.debug(e) + with open(os.path.join( + self._root_dir, PROFILES_DIR, risk_profile_file + ), encoding='utf-8') as f: + json_data = json.load(f) + risk_profile = RiskProfile() + risk_profile = risk_profile.load( + profile_json=json_data, + profile_format=self._profile_format) + self._profiles.append(risk_profile) + + # except Exception as e: + # LOGGER.error('An error occurred whilst loading risk profiles') + # LOGGER.debug(e) def get_profiles_format(self): return self._profile_format_json @@ -398,94 +430,37 @@ def get_profile(self, name): return profile return None - def validate_profile(self, profile_json): - - # Check name field is present - if 'name' not in profile_json: - return False - - # Check questions field is present - if 'questions' not in profile_json: - return False - - # Check all questions are present - for format_q in self.get_profiles_format(): - if self._get_profile_question( - profile_json, format_q.get('question')) is None: - LOGGER.error('Missing question: ' + format_q.get('question')) - return False - - return True - - def _get_profile_question(self, profile_json, question): - - for q in profile_json.get('questions'): - if question.lower() == q.get('question').lower(): - return q - - return None - def update_profile(self, profile_json): profile_name = profile_json['name'] # Add version, timestamp and status profile_json['version'] = self.get_version() - profile_json['created'] = datetime.datetime.now().strftime('%Y-%m-%d') - - if 'status' in profile_json and profile_json.get('status') == 'Valid': - # Attempting to submit a risk profile, we need to check it - - # Check all questions have been answered - all_questions_answered = True - - for question in self.get_profiles_format(): - - # Check question is present - profile_question = self._get_profile_question( - profile_json, question.get('question') - ) - - if profile_question is not None: - - # Check answer is present - if 'answer' not in profile_question: - LOGGER.error( - 'Missing answer for question: ' + question.get('question')) - all_questions_answered = False - - else: - LOGGER.error('Missing question: ' + question.get('question')) - all_questions_answered = False - - if not all_questions_answered: - LOGGER.error('Not all questions answered') - return None - - else: - profile_json['status'] = 'Draft' risk_profile = self.get_profile(profile_name) if risk_profile is None: # Create a new risk profile - risk_profile = RiskProfile(profile_json) + risk_profile = RiskProfile(profile_json=profile_json, + profile_format=self._profile_format) self._profiles.append(risk_profile) else: + risk_profile.update(profile_json, self._profile_format) # Check if name has changed if 'rename' in profile_json: - new_name = profile_json.get('rename') - # Delete the original file os.remove(os.path.join(PROFILES_DIR, risk_profile.name + '.json')) - risk_profile.name = new_name + # Find the index of the risk_profile to replace + index_to_replace = next( + (index for (index, d) in enumerate( + self._profiles) if d.name == profile_name), None) - # Update questions and answers - risk_profile.questions = profile_json.get('questions') + if index_to_replace is not None: + self._profiles[index_to_replace] = risk_profile # Write file to disk with open( @@ -493,24 +468,10 @@ def update_profile(self, profile_json): PROFILES_DIR, risk_profile.name + '.json' ), 'w', encoding='utf-8') as f: - f.write(json.dumps(risk_profile.to_json())) + f.write(risk_profile.to_json()) return risk_profile - def check_profile_status(self, profile): - - if profile.status == 'Valid': - - # Check expiry - created_date = profile.created.timestamp() - - today = datetime.datetime.now().timestamp() - - if created_date < (today - SECONDS_IN_YEAR): - profile.status = 'Expired' - - return profile.status - def delete_profile(self, profile): try: diff --git a/framework/python/src/test_orc/test_orchestrator.py b/framework/python/src/test_orc/test_orchestrator.py index 15b933a1c..d3c12d905 100644 --- a/framework/python/src/test_orc/test_orchestrator.py +++ b/framework/python/src/test_orc/test_orchestrator.py @@ -123,7 +123,7 @@ def run_test_modules(self): self.get_session().set_report_url(report.get_report_url()) # Move testing output from runtime to local device folder - timestamp_dir = self._timestamp_results(device) + self._timestamp_results(device) LOGGER.debug("Cleaning old test results...") self._cleanup_old_test_results(device) diff --git a/framework/requirements.txt b/framework/requirements.txt index 3989c3b74..ee96c7e61 100644 --- a/framework/requirements.txt +++ b/framework/requirements.txt @@ -26,4 +26,7 @@ markdown==3.5.2 # Requirements for the session cryptography==42.0.7 -pytz==2024.1 \ No newline at end of file +pytz==2024.1 + +# Requirements for the risk profile +python-dateutil==2.9.0 \ No newline at end of file diff --git a/local/.gitignore b/local/.gitignore index 06f79c1ca..84d72ff6e 100644 --- a/local/.gitignore +++ b/local/.gitignore @@ -1,3 +1,4 @@ system.json devices root_certs +risk_profiles diff --git a/local/risk_profiles/Primary profile.json b/local/risk_profiles/Primary profile.json new file mode 100644 index 000000000..f95947111 --- /dev/null +++ b/local/risk_profiles/Primary profile.json @@ -0,0 +1 @@ +{"name": "Primary profile", "version": "1.3-alpha", "created": "2024-06-27", "status": "Valid", "risk": "High", "questions": [{"question": "What type of device is this?", "answer": "IoT Gateway", "risk": "High"}, {"question": "How will this device be used at Google?", "answer": "Hey"}, {"question": "Is this device going to be managed by Google or a third party?", "answer": "Google", "risk": "Limited"}, {"question": "Will the third-party device administrator be able to grant access to authorized Google personnel upon request?", "answer": "Yes", "risk": "Limited"}, {"question": "Are any of the following statements true about your device?", "answer": [3], "risk": "Limited"}, {"question": "Which of the following statements are true about this device?", "answer": [5], "risk": "Limited"}, {"question": "Does the network protocol assure server-to-client identity verification?", "answer": "Yes", "risk": "Limited"}, {"question": "Click the statements that best describe the characteristics of this device.", "answer": [4], "risk": "High"}, {"question": "Are any of the following statements true about this device?", "answer": [0], "risk": "High"}]} \ No newline at end of file diff --git a/modules/test/base/python/src/test_module.py b/modules/test/base/python/src/test_module.py index dc924cfeb..200a29a2b 100644 --- a/modules/test/base/python/src/test_module.py +++ b/modules/test/base/python/src/test_module.py @@ -50,7 +50,7 @@ def _add_logger(self, log_name, module_name, log_dir=None): global LOGGER LOGGER = logger.get_logger(name=log_name, log_file=module_name, - log_dir=log_dir) + log_dir=log_dir) # pylint: disable=E1123 def generate_module_report(self): pass diff --git a/resources/risk_assessment.json b/resources/risk_assessment.json index 5393edfbd..784322421 100644 --- a/resources/risk_assessment.json +++ b/resources/risk_assessment.json @@ -3,10 +3,170 @@ "question": "What type of device is this?", "type": "select", "options": [ - "IoT Sensor", - "IoT Controller", - "Smart Device", - "Something else" + { + "text": "Building Automation Gateway", + "risk": "High" + }, + { + "text": "IoT Gateway", + "risk": "High" + }, + { + "text": "Controller - AHU", + "risk": "Limited" + }, + { + "text": "Controller - Boiler", + "risk": "High" + }, + { + "text": "Controller - Chiller", + "risk": "High" + }, + { + "text": "Controller - FCU", + "risk": "Limited" + }, + { + "text": "Controller - Pump", + "risk": "Limited" + }, + { + "text": "Controller - CRAC", + "risk": "Limited" + }, + { + "text": "Controller - VAV", + "risk": "Limited" + }, + { + "text": "Controller - VRF", + "risk": "Limited" + }, + { + "text": "Controller - Multiple", + "risk": "High" + }, + { + "text": "Controller - Other", + "risk": "High" + }, + { + "text": "Controller - Lighting", + "risk": "Limited" + }, + { + "text": "Controller - Blinds/Facades", + "risk": "Limited" + }, + { + "text": "Controller - Lifts/Elevators", + "risk": "High" + }, + { + "text": "Controller - UPS", + "risk": "Limited" + }, + { + "text": "Sensor - Air Quality", + "risk": "Limited" + }, + { + "text": "Sensor - Vibration", + "risk": "Limited" + }, + { + "text": "Sensor - Humidity", + "risk": "Limited" + }, + { + "text": "Sensor - Water", + "risk": "Limited" + }, + { + "text": "Sensor - Occupancy", + "risk": "High" + }, + { + "text": "Sensor - Volume", + "risk": "Limited" + }, + { + "text": "Sensor - Weight", + "risk": "Limited" + }, + { + "text": "Sensor - Weather", + "risk": "Limited" + }, + { + "text": "Sensor - Steam", + "risk": "High" + }, + { + "text": "Sensor - Air Flow", + "risk": "Limited" + }, + { + "text": "Sensor - Lighting", + "risk": "Limited" + }, + { + "text": "Sensor - Other", + "risk": "High" + }, + { + "text": "Sensor - Air Quality", + "risk": "Limited" + }, + { + "text": "Monitoring - Fire System", + "risk": "Limited" + }, + { + "text": "Monitoring - Emergency Lighting", + "risk": "Limited" + }, + { + "text": "Monitoring - Other", + "risk": "High" + }, + { + "text": "Monitoring - UPS", + "risk": "Limited" + }, + { + "text": "Meter - Water", + "risk": "Limited" + }, + { + "text": "Meter - Gas", + "risk": "Limited" + }, + { + "text": "Meter - Electricity", + "risk": "Limited" + }, + { + "text": "Meter - Other", + "risk": "High" + }, + { + "text": "Other", + "risk": "High" + }, + { + "text": "Data - Storage", + "risk": "High" + }, + { + "text": "Data - Processing", + "risk": "High" + }, + { + "text": "Tablet", + "risk": "Limited" + } ], "validation": { "required": true @@ -14,26 +174,26 @@ }, { "question": "How will this device be used at Google?", + "description": "Desribe your use case. Add links to user journey diagrams and TDD if available.", "type": "text-long", "validation": { - "max": "128", + "max": "512", "required": true } }, - { - "question": "What is the email of the device owner(s)?", - "type": "email-multiple", - "validation": { - "required": true, - "max": "128" - } - }, { "question": "Is this device going to be managed by Google or a third party?", + "description": "A manufacturer or supplier is considered third party in this case", "type": "select", "options": [ - "Google", - "Third Party" + { + "text": "Google", + "risk": "Limited" + }, + { + "text": "Third Party", + "risk": "High" + } ], "validation": { "required": true @@ -43,9 +203,15 @@ "question": "Will the third-party device administrator be able to grant access to authorized Google personnel upon request?", "type": "select", "options": [ - "Yes", - "No", - "N/A" + { + "text": "Yes" + }, + { + "text": "No" + }, + { + "text": "N/A" + } ], "default": "N/A", "validation": { @@ -53,76 +219,159 @@ } }, { + "category": "Data Collection", "question": "Are any of the following statements true about your device?", "description": "This tells us about the data your device will collect", "type": "select-multiple", "options": [ - "The device collects any Personal Identifiable Information (PII) or Personal Health Information (PHI)", - "The device collects intellectual property and trade secrets, sensitive business data, critical infrastructure data, identity assets", - "The device stream confidential business data in real-time (seconds)?", - "None of the above" + { + "text": "The device collects any Personal Identifiable Information (PII) or Personal Health Information (PHI)", + "risk": "High" + }, + { + "text": "The device collects intellectual property and trade secrets, sensitive business data, critical infrastructure data, identity assets", + "risk": "High" + }, + { + "text": "The device streams confidential business data in real-time (seconds)?", + "risk": "High" + }, + { + "text": "None of the above", + "risk": "Limited" + } ], "validation": { "required": true } }, { + "category": "Data Transmission", "question": "Which of the following statements are true about this device?", "description": "This tells us about the types of data that are transmitted from this device and how the transmission is performed from a technical standpoint.", "type": "select-multiple", "options": [ - "PII/PHI, confidential business data, or crown jewel data is transmitted to a destination outside Alphabet's ownership", - "Data transmission occurs across less-trusted networks (e.g. the internet).", - "A failure in data transmission would likely have a substantial negative impact (https://www.rra.rocks/docs/standard_levels#levels-definitions)", - "A confidentiality breach during transmission would have a substantial negative impact", - "The device encrypts data during transmission", - "The device network protocol is well-established and currently used by Google", - "None of the above" + { + "text": "PII/PHI, confidential business data, or crown jewel data is transmitted to a destination outside Alphabet's ownership", + "risk": "High" + }, + { + "text": "Data transmission occurs across less-trusted networks (e.g. the internet).", + "risk": "High" + }, + { + "text": "A failure in data transmission would likely have a substantial negative impact (https://www.rra.rocks/docs/standard_levels#levels-definitions)", + "risk": "High" + }, + { + "text": "A confidentiality breach during transmission would have a substantial negative impact", + "risk": "High" + }, + { + "text": "The device does not encrypt data during transmission", + "risk": "High" + }, + { + "text": "None of the above", + "risk": "Limited" + } ], "validation": { "required": true } }, { + "category": "Data Transmission", "question": "Does the network protocol assure server-to-client identity verification?", "type": "select", "options": [ - "Yes", - "No", - "I don't know" + { + "text": "Yes", + "risk": "Limited" + }, + { + "text": "No", + "risk": "High" + }, + { + "text": "I don't know", + "risk": "High" + } + ], "validation": { "required": true } }, { + "category": "Remote Operation", "question": "Click the statements that best describe the characteristics of this device.", "description": "This tells us about how this device is managed remotely.", "type": "select-multiple", "options": [ - "PII/PHI, or confidential business data is accessible from the device without authentication", - "Unrecoverable actions (e.g. disk wipe) can be performed remotely", - "Authentication is required for remote access", - "The management interface is accessible from the public internet", - "Static credentials are used for administration", - "None of the above" + { + "text": "PII/PHI, or confidential business data is accessible from the device without authentication", + "risk": "High" + }, + { + "text": "Unrecoverable actions (e.g. disk wipe) can be performed remotely", + "risk": "High" + }, + { + "text": "Authentication is not required for remote access", + "risk": "High" + }, + { + "text": "The management interface is accessible from the public internet", + "risk": "High" + }, + { + "text": "Static credentials are used for administration", + "risk": "High" + }, + { + "text": "None of the above", + "risk": "Limited" + } ], "validation": { "required": true } }, { + "category": "Operating Environment", "question": "Are any of the following statements true about this device?", "description": "This informs us about what other systems and processes this device is a part of.", "type": "select-multiple", "options": [ - "The device monitors an environment for active risks to human life.", - "The device is used to convey people, or critical property.", - "The device controls robotics in human-accessible spaces.", - "The device controls physical access systems.", - "The device is involved in processes required by regulations, or compliance. (ex. privacy, security, safety regulations)", - "The device's failure would cause faults in other high-criticality processes.", - "None of the above" + { + "text": "The device monitors an environment for active risks to human life.", + "risk": "High" + }, + { + "text": "The device is used to convey people, or critical property.", + "risk": "High" + }, + { + "text": "The device controls robotics in human-accessible spaces.", + "risk": "High" + }, + { + "text": "The device controls physical access systems.", + "risk": "High" + }, + { + "text": "The device is involved in processes required by regulations, or compliance. (ex. privacy, security, safety regulations)", + "risk": "High" + }, + { + "text": "The device's failure would cause faults in other high-criticality processes.", + "risk": "High" + }, + { + "text": "None of the above", + "risk": "Limited" + } ], "validation": { "required": true diff --git a/testing/tests/test_tests.py b/testing/tests/test_tests.py index 376dad333..aaae1a09d 100644 --- a/testing/tests/test_tests.py +++ b/testing/tests/test_tests.py @@ -79,20 +79,20 @@ def test_list_tests(capsys, results, test_matrix): itertools.chain.from_iterable( [collect_actual_results(results[x]) for x in results.keys()])) - ci_pass = set([ + ci_pass = set( test for testers in test_matrix.values() for test, result in testers['expected_results'].items() if result == 'Compliant' - ]) + ) - ci_fail = set([ + ci_fail = set( test for testers in test_matrix.values() for test, result in testers['expected_results'].items() if result == 'Non-Compliant' - ]) + ) with capsys.disabled(): - #TODO print matching the JSON schema for easy copy/paste + # TODO: print matching the JSON schema for easy copy/paste print('============') print('============') print('tests seen:') diff --git a/testing/unit/risk_profile/profiles/risk_profile_draft.json b/testing/unit/risk_profile/profiles/risk_profile_draft.json new file mode 100644 index 000000000..315ec3417 --- /dev/null +++ b/testing/unit/risk_profile/profiles/risk_profile_draft.json @@ -0,0 +1,144 @@ +{ + "name": "Primary profile", + "status": "Valid", + "created": "2024-05-23", + "version": "v1.3", + "questions": [ + { + "question": "What type of device is this?", + "type": "select", + "options": [ + "IoT Sensor", + "IoT Controller", + "Smart Device", + "Something else" + ], + "answer": "IoT Sensor", + "validation": { + "required": true + } + }, + { + "question": "How will this device be used at Google?", + "type": "text-long", + "answer": "Installed in a building", + "validation": { + "max": "128", + "required": true + } + }, + { + "question": "What is the email of the device owner(s)?", + "type": "email-multiple", + "answer": "boddey@google.com, cmeredith@google.com", + "validation": { + "required": true, + "max": "128" + } + }, + { + "question": "Is this device going to be managed by Google or a third party?", + "type": "select", + "options": [ + "Google", + "Third Party" + ], + "answer": "Google", + "validation": { + "required": true + } + }, + { + "question": "Will the third-party device administrator be able to grant access to authorized Google personnel upon request?", + "type": "select", + "options": [ + "Yes", + "No", + "N/A" + ], + "default": "N/A", + "answer": "Yes", + "validation": { + "required": true + } + }, + { + "category": "Data Transmission", + "question": "Which of the following statements are true about this device?", + "description": "This tells us about the types of data that are transmitted from this device and how the transmission is performed from a technical standpoint.", + "type": "select-multiple", + "answer": [ + 0, + 1, + 5 + ], + "options": [ + "PII/PHI, confidential business data, or crown jewel data is transmitted to a destination outside Alphabet's ownership", + "Data transmission occurs across less-trusted networks (e.g. the internet).", + "A failure in data transmission would likely have a substantial negative impact (https://www.rra.rocks/docs/standard_levels#levels-definitions)", + "A confidentiality breach during transmission would have a substantial negative impact", + "The device encrypts data during transmission", + "The device network protocol is well-established and currently used by Google" + ], + "validation": { + "required": true + } + }, + { + "category": "Data Transmission", + "question": "Does the network protocol assure server-to-client identity verification?", + "type": "select", + "answer": "Yes", + "options": [ + "Yes", + "No", + "I don't know" + ], + "validation": { + "required": true + } + }, + { + "category": "Remote Operation", + "question": "Click the statements that best describe the characteristics of this device.", + "description": "This tells us about how this device is managed remotely.", + "type": "select-multiple", + "answer": [ + 0, + 1, + 2 + ], + "options": [ + "PII/PHI, or confidential business data is accessible from the device without authentication", + "Unrecoverable actions (e.g. disk wipe) can be performed remotely", + "Authentication is required for remote access", + "The management interface is accessible from the public internet", + "Static credentials are used for administration" + ], + "validation": { + "required": true + } + }, + { + "category": "Operating Environment", + "question": "Are any of the following statements true about this device?", + "description": "This informs us about what other systems and processes this device is a part of.", + "type": "select-multiple", + "answer": [ + 2, + 3 + ], + "options": [ + "The device monitors an environment for active risks to human life.", + "The device is used to convey people, or critical property.", + "The device controls robotics in human-accessible spaces.", + "The device controls physical access systems.", + "The device is involved in processes required by regulations, or compliance. (ex. privacy, security, safety regulations)", + "The device's failure would cause faults in other high-criticality processes." + ], + "validation": { + "required": true + } + } + ] +} \ No newline at end of file diff --git a/testing/unit/risk_profile/profiles/risk_profile_expired.json b/testing/unit/risk_profile/profiles/risk_profile_expired.json new file mode 100644 index 000000000..5bcedab07 --- /dev/null +++ b/testing/unit/risk_profile/profiles/risk_profile_expired.json @@ -0,0 +1,162 @@ +{ + "name": "Primary profile", + "status": "Valid", + "created": "2022-05-23", + "version": "v1.3", + "questions": [ + { + "question": "What type of device is this?", + "type": "select", + "options": [ + "IoT Sensor", + "IoT Controller", + "Smart Device", + "Something else" + ], + "answer": "IoT Sensor", + "validation": { + "required": true + } + }, + { + "question": "How will this device be used at Google?", + "type": "text-long", + "answer": "Installed in a building", + "validation": { + "max": "128", + "required": true + } + }, + { + "question": "What is the email of the device owner(s)?", + "type": "email-multiple", + "answer": "boddey@google.com, cmeredith@google.com", + "validation": { + "required": true, + "max": "128" + } + }, + { + "question": "Is this device going to be managed by Google or a third party?", + "type": "select", + "options": [ + "Google", + "Third Party" + ], + "answer": "Google", + "validation": { + "required": true + } + }, + { + "question": "Will the third-party device administrator be able to grant access to authorized Google personnel upon request?", + "type": "select", + "options": [ + "Yes", + "No", + "N/A" + ], + "default": "N/A", + "answer": "Yes", + "validation": { + "required": true + } + }, + { + "category": "Data Collection", + "question": "Are any of the following statements true about your device?", + "description": "This tells us about the data your device will collect", + "type": "select-multiple", + "answer": [ + 0, + 2 + ], + "options": [ + "The device collects any Personal Identifiable Information (PII) or Personal Health Information (PHI)", + "The device collects intellectual property and trade secrets, sensitive business data, critical infrastructure data, identity assets", + "The device stream confidential business data in real-time (seconds)?" + ], + "validation": { + "required": true + } + }, + { + "category": "Data Transmission", + "question": "Which of the following statements are true about this device?", + "description": "This tells us about the types of data that are transmitted from this device and how the transmission is performed from a technical standpoint.", + "type": "select-multiple", + "answer": [ + 0, + 1, + 5 + ], + "options": [ + "PII/PHI, confidential business data, or crown jewel data is transmitted to a destination outside Alphabet's ownership", + "Data transmission occurs across less-trusted networks (e.g. the internet).", + "A failure in data transmission would likely have a substantial negative impact (https://www.rra.rocks/docs/standard_levels#levels-definitions)", + "A confidentiality breach during transmission would have a substantial negative impact", + "The device encrypts data during transmission", + "The device network protocol is well-established and currently used by Google" + ], + "validation": { + "required": true + } + }, + { + "category": "Data Transmission", + "question": "Does the network protocol assure server-to-client identity verification?", + "type": "select", + "answer": "Yes", + "options": [ + "Yes", + "No", + "I don't know" + ], + "validation": { + "required": true + } + }, + { + "category": "Remote Operation", + "question": "Click the statements that best describe the characteristics of this device.", + "description": "This tells us about how this device is managed remotely.", + "type": "select-multiple", + "answer": [ + 0, + 1, + 2 + ], + "options": [ + "PII/PHI, or confidential business data is accessible from the device without authentication", + "Unrecoverable actions (e.g. disk wipe) can be performed remotely", + "Authentication is required for remote access", + "The management interface is accessible from the public internet", + "Static credentials are used for administration" + ], + "validation": { + "required": true + } + }, + { + "category": "Operating Environment", + "question": "Are any of the following statements true about this device?", + "description": "This informs us about what other systems and processes this device is a part of.", + "type": "select-multiple", + "answer": [ + 2, + 3 + ], + "options": [ + "The device monitors an environment for active risks to human life.", + "The device is used to convey people, or critical property.", + "The device controls robotics in human-accessible spaces.", + "The device controls physical access systems.", + "The device is involved in processes required by regulations, or compliance. (ex. privacy, security, safety regulations)", + "The device's failure would cause faults in other high-criticality processes." + ], + "validation": { + "required": true + } + } + ] +} \ No newline at end of file diff --git a/testing/unit/risk_profile/profiles/risk_profile_valid_high.json b/testing/unit/risk_profile/profiles/risk_profile_valid_high.json new file mode 100644 index 000000000..78ed9a5fa --- /dev/null +++ b/testing/unit/risk_profile/profiles/risk_profile_valid_high.json @@ -0,0 +1,162 @@ +{ + "name": "Primary profile", + "status": "Valid", + "created": "2024-05-23", + "version": "v1.3", + "questions": [ + { + "question": "What type of device is this?", + "type": "select", + "options": [ + "IoT Sensor", + "IoT Controller", + "Smart Device", + "Something else" + ], + "answer": "IoT Sensor", + "validation": { + "required": true + } + }, + { + "question": "How will this device be used at Google?", + "type": "text-long", + "answer": "Installed in a building", + "validation": { + "max": "128", + "required": true + } + }, + { + "question": "What is the email of the device owner(s)?", + "type": "email-multiple", + "answer": "boddey@google.com, cmeredith@google.com", + "validation": { + "required": true, + "max": "128" + } + }, + { + "question": "Is this device going to be managed by Google or a third party?", + "type": "select", + "options": [ + "Google", + "Third Party" + ], + "answer": "Google", + "validation": { + "required": true + } + }, + { + "question": "Will the third-party device administrator be able to grant access to authorized Google personnel upon request?", + "type": "select", + "options": [ + "Yes", + "No", + "N/A" + ], + "default": "N/A", + "answer": "Yes", + "validation": { + "required": true + } + }, + { + "category": "Data Collection", + "question": "Are any of the following statements true about your device?", + "description": "This tells us about the data your device will collect", + "type": "select-multiple", + "answer": [ + 0, + 2 + ], + "options": [ + "The device collects any Personal Identifiable Information (PII) or Personal Health Information (PHI)", + "The device collects intellectual property and trade secrets, sensitive business data, critical infrastructure data, identity assets", + "The device stream confidential business data in real-time (seconds)?" + ], + "validation": { + "required": true + } + }, + { + "category": "Data Transmission", + "question": "Which of the following statements are true about this device?", + "description": "This tells us about the types of data that are transmitted from this device and how the transmission is performed from a technical standpoint.", + "type": "select-multiple", + "answer": [ + 0, + 1, + 5 + ], + "options": [ + "PII/PHI, confidential business data, or crown jewel data is transmitted to a destination outside Alphabet's ownership", + "Data transmission occurs across less-trusted networks (e.g. the internet).", + "A failure in data transmission would likely have a substantial negative impact (https://www.rra.rocks/docs/standard_levels#levels-definitions)", + "A confidentiality breach during transmission would have a substantial negative impact", + "The device encrypts data during transmission", + "The device network protocol is well-established and currently used by Google" + ], + "validation": { + "required": true + } + }, + { + "category": "Data Transmission", + "question": "Does the network protocol assure server-to-client identity verification?", + "type": "select", + "answer": "Yes", + "options": [ + "Yes", + "No", + "I don't know" + ], + "validation": { + "required": true + } + }, + { + "category": "Remote Operation", + "question": "Click the statements that best describe the characteristics of this device.", + "description": "This tells us about how this device is managed remotely.", + "type": "select-multiple", + "answer": [ + 0, + 1, + 2 + ], + "options": [ + "PII/PHI, or confidential business data is accessible from the device without authentication", + "Unrecoverable actions (e.g. disk wipe) can be performed remotely", + "Authentication is required for remote access", + "The management interface is accessible from the public internet", + "Static credentials are used for administration" + ], + "validation": { + "required": true + } + }, + { + "category": "Operating Environment", + "question": "Are any of the following statements true about this device?", + "description": "This informs us about what other systems and processes this device is a part of.", + "type": "select-multiple", + "answer": [ + 2, + 3 + ], + "options": [ + "The device monitors an environment for active risks to human life.", + "The device is used to convey people, or critical property.", + "The device controls robotics in human-accessible spaces.", + "The device controls physical access systems.", + "The device is involved in processes required by regulations, or compliance. (ex. privacy, security, safety regulations)", + "The device's failure would cause faults in other high-criticality processes." + ], + "validation": { + "required": true + } + } + ] +} \ No newline at end of file diff --git a/testing/unit/risk_profile/profiles/risk_profile_valid_limited.json b/testing/unit/risk_profile/profiles/risk_profile_valid_limited.json new file mode 100644 index 000000000..aa6c73297 --- /dev/null +++ b/testing/unit/risk_profile/profiles/risk_profile_valid_limited.json @@ -0,0 +1,148 @@ +{ + "name": "Primary profile", + "status": "Valid", + "created": "2024-05-23", + "version": "v1.3", + "questions": [ + { + "question": "What type of device is this?", + "type": "select", + "options": [ + "IoT Sensor", + "IoT Controller", + "Smart Device", + "Something else" + ], + "answer": "IoT Sensor", + "validation": { + "required": true + } + }, + { + "question": "How will this device be used at Google?", + "type": "text-long", + "answer": "Installed in a building", + "validation": { + "max": "128", + "required": true + } + }, + { + "question": "What is the email of the device owner(s)?", + "type": "email-multiple", + "answer": "boddey@google.com, cmeredith@google.com", + "validation": { + "required": true, + "max": "128" + } + }, + { + "question": "Is this device going to be managed by Google or a third party?", + "type": "select", + "options": [ + "Google", + "Third Party" + ], + "answer": "Google", + "validation": { + "required": true + } + }, + { + "question": "Will the third-party device administrator be able to grant access to authorized Google personnel upon request?", + "type": "select", + "options": [ + "Yes", + "No", + "N/A" + ], + "default": "N/A", + "answer": "Yes", + "validation": { + "required": true + } + }, + { + "category": "Data Collection", + "question": "Are any of the following statements true about your device?", + "description": "This tells us about the data your device will collect", + "type": "select-multiple", + "answer": [], + "options": [ + "The device collects any Personal Identifiable Information (PII) or Personal Health Information (PHI)", + "The device collects intellectual property and trade secrets, sensitive business data, critical infrastructure data, identity assets", + "The device stream confidential business data in real-time (seconds)?" + ], + "validation": { + "required": true + } + }, + { + "category": "Data Transmission", + "question": "Which of the following statements are true about this device?", + "description": "This tells us about the types of data that are transmitted from this device and how the transmission is performed from a technical standpoint.", + "type": "select-multiple", + "answer": [], + "options": [ + "PII/PHI, confidential business data, or crown jewel data is transmitted to a destination outside Alphabet's ownership", + "Data transmission occurs across less-trusted networks (e.g. the internet).", + "A failure in data transmission would likely have a substantial negative impact (https://www.rra.rocks/docs/standard_levels#levels-definitions)", + "A confidentiality breach during transmission would have a substantial negative impact", + "The device encrypts data during transmission", + "The device network protocol is well-established and currently used by Google" + ], + "validation": { + "required": true + } + }, + { + "category": "Data Transmission", + "question": "Does the network protocol assure server-to-client identity verification?", + "type": "select", + "answer": "Yes", + "options": [ + "Yes", + "No", + "I don't know" + ], + "validation": { + "required": true + } + }, + { + "category": "Remote Operation", + "question": "Click the statements that best describe the characteristics of this device.", + "description": "This tells us about how this device is managed remotely.", + "type": "select-multiple", + "answer": [], + "options": [ + "PII/PHI, or confidential business data is accessible from the device without authentication", + "Unrecoverable actions (e.g. disk wipe) can be performed remotely", + "Authentication is required for remote access", + "The management interface is accessible from the public internet", + "Static credentials are used for administration" + ], + "validation": { + "required": true + } + }, + { + "category": "Operating Environment", + "question": "Are any of the following statements true about this device?", + "description": "This informs us about what other systems and processes this device is a part of.", + "type": "select-multiple", + "answer": [], + "options": [ + "The device monitors an environment for active risks to human life.", + "The device is used to convey people, or critical property.", + "The device controls robotics in human-accessible spaces.", + "The device controls physical access systems.", + "The device is involved in processes required by regulations, or compliance. (ex. privacy, security, safety regulations)", + "The device's failure would cause faults in other high-criticality processes." + ], + "validation": { + "required": true + } + } + ] +} \ No newline at end of file diff --git a/testing/unit/risk_profile/risk_profile_test.py b/testing/unit/risk_profile/risk_profile_test.py new file mode 100644 index 000000000..c4c80c1e4 --- /dev/null +++ b/testing/unit/risk_profile/risk_profile_test.py @@ -0,0 +1,139 @@ +# Copyright 2023 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Module run all the Risk Profile related unit tests""" +import unittest +import os +import json +from risk_profile import RiskProfile + +SECONDS_IN_YEAR = 31536000 + +MODULE = 'risk_profile' + +# Define the file paths +UNIT_TEST_DIR = 'testing/unit/' +TEST_FILES_DIR = os.path.join('testing/unit', MODULE) +PROFILE_DIR = os.path.join(TEST_FILES_DIR, 'profiles') +OUTPUT_DIR = os.path.join(TEST_FILES_DIR, 'output/') + + +class RiskProfileTest(unittest.TestCase): + """Contains and runs all the unit tests concerning DNS behaviors""" + + @classmethod + def setUpClass(cls): + # Create the output directories and ignore errors if it already exists + os.makedirs(OUTPUT_DIR, exist_ok=True) + with open('resources/risk_assessment.json', 'r', encoding='utf-8') as file: + cls.profile_format = json.loads(file.read()) + + def risk_profile_high_test(self): + # Read the risk profile json file + risk_profile_path = os.path.join(PROFILE_DIR, + 'risk_profile_valid_high.json') + with open(risk_profile_path, 'r', encoding='utf-8') as file: + risk_profile_json = json.loads(file.read()) + + # Create the RiskProfile object from the json file + risk_profile = RiskProfile(risk_profile_json, self.profile_format) + + # Write the profile to file + output_file = os.path.join(OUTPUT_DIR, 'risk_profile_high.json') + with open(output_file, 'w', encoding='utf-8') as file: + file.write(risk_profile.to_json(pretty=True)) + + self.assertEqual(risk_profile.risk, 'High') + + def risk_profile_limited_test(self): + # Read the risk profile json file + risk_profile_path = os.path.join(PROFILE_DIR, + 'risk_profile_valid_limited.json') + with open(risk_profile_path, 'r', encoding='utf-8') as file: + risk_profile_json = json.loads(file.read()) + + # Create the RiskProfile object from the json file + risk_profile = RiskProfile(risk_profile_json, self.profile_format) + + # Write the profile to file + output_file = os.path.join(OUTPUT_DIR, 'risk_profile_limited.json') + with open(output_file, 'w', encoding='utf-8') as file: + file.write(risk_profile.to_json(pretty=True)) + + self.assertEqual(risk_profile.risk, 'Limited') + + def risk_profile_rename_test(self): + # Read the risk profile json file + risk_profile_path = os.path.join(PROFILE_DIR, + 'risk_profile_valid_high.json') + with open(risk_profile_path, 'r', encoding='utf-8') as file: + risk_profile_json = json.loads(file.read()) + + + # Create the RiskProfile object from the json file + risk_profile = RiskProfile(risk_profile_json, self.profile_format) + + # Rename the profile + risk_profile_json['rename'] = 'Primary profile renamed' + risk_profile.update(risk_profile_json, self.profile_format) + + # Write the renamed profile to file + output_file = os.path.join(OUTPUT_DIR, 'risk_profile_renamed.json') + with open(output_file, 'w', encoding='utf-8') as file: + file.write(risk_profile.to_json(pretty=True)) + + self.assertEqual(risk_profile.name, 'Primary profile renamed') + + def risk_profile_draft_test(self): + # Read the risk profile json file + risk_profile_path = os.path.join(PROFILE_DIR, 'risk_profile_draft.json') + with open(risk_profile_path, 'r', encoding='utf-8') as file: + risk_profile_json = json.loads(file.read()) + + # Create the RiskProfile object from the json file + risk_profile = RiskProfile(risk_profile_json, self.profile_format) + + # Write the profile to file + output_file = os.path.join(OUTPUT_DIR, 'risk_profile_draft.json') + with open(output_file, 'w', encoding='utf-8') as file: + file.write(risk_profile.to_json(pretty=True)) + + self.assertEqual(risk_profile.status, 'Draft') + + def risk_profile_expired_test(self): + # Read the risk profile json file + risk_profile_path = os.path.join(PROFILE_DIR, 'risk_profile_expired.json') + with open(risk_profile_path, 'r', encoding='utf-8') as file: + risk_profile_json = json.loads(file.read()) + + # Create the RiskProfile object from the json file + risk_profile = RiskProfile().load(risk_profile_json, self.profile_format) + + # Write the profile to file + output_file = os.path.join(OUTPUT_DIR, 'risk_profile_expired.json') + with open(output_file, 'w', encoding='utf-8') as file: + file.write(risk_profile.to_json(pretty=True)) + + self.assertEqual(risk_profile.status, 'Expired') + +if __name__ == '__main__': + suite = unittest.TestSuite() + + suite.addTest(RiskProfileTest('risk_profile_high_test')) + suite.addTest(RiskProfileTest('risk_profile_limited_test')) + suite.addTest(RiskProfileTest('risk_profile_rename_test')) + suite.addTest(RiskProfileTest('risk_profile_draft_test')) + suite.addTest(RiskProfileTest('risk_profile_expired_test')) + + runner = unittest.TextTestRunner() + runner.run(suite) diff --git a/testing/unit/run_tests.sh b/testing/unit/run_tests.sh index 726019b23..975627fc3 100644 --- a/testing/unit/run_tests.sh +++ b/testing/unit/run_tests.sh @@ -30,32 +30,35 @@ PYTHONPATH="$PYTHONPATH:$PWD/modules/test/base/python/src" PYTHONPATH="$PYTHONPATH:$PWD/modules/test/conn/python/src" PYTHONPATH="$PYTHONPATH:$PWD/modules/test/tls/python/src" PYTHONPATH="$PYTHONPATH:$PWD/modules/test/dns/python/src" -PYTHONPATH="$PYTHONPATH:$PWD/modules/test/nmap/python/src" +PYTHONPATH="$PYTHONPATH:$PWD/modules/test/services/python/src" PYTHONPATH="$PYTHONPATH:$PWD/modules/test/ntp/python/src" # Set the python path with all sources export PYTHONPATH -# # Run the DHCP Unit tests -# python3 -u $PWD/modules/network/dhcp-1/python/src/grpc_server/dhcp_config_test.py -# python3 -u $PWD/modules/network/dhcp-2/python/src/grpc_server/dhcp_config_test.py +# Run the DHCP Unit tests +python3 -u $PWD/modules/network/dhcp-1/python/src/grpc_server/dhcp_config_test.py +python3 -u $PWD/modules/network/dhcp-2/python/src/grpc_server/dhcp_config_test.py -# # Run the Conn Module Unit Tests -# python3 -u $PWD/testing/unit/conn/conn_module_test.py +# Run the Conn Module Unit Tests +python3 -u $PWD/testing/unit/conn/conn_module_test.py # Run the TLS Module Unit Tests python3 -u $PWD/testing/unit/tls/tls_module_test.py -# # Run the DNS Module Unit Tests -# python3 -u $PWD/testing/unit/dns/dns_module_test.py +# Run the DNS Module Unit Tests +python3 -u $PWD/testing/unit/dns/dns_module_test.py -# # Run the NMAP Module Unit Tests -# python3 -u $PWD/testing/unit/nmap/nmap_module_test.py +# Run the NMAP Module Unit Tests +python3 -u $PWD/testing/unit/services/services_module_test.py -# # Run the NTP Module Unit Tests -# python3 -u $PWD/testing/unit/ntp/ntp_module_test.py +# Run the NTP Module Unit Tests +python3 -u $PWD/testing/unit/ntp/ntp_module_test.py -# # # Run the Report Unit Tests -# python3 -u $PWD/testing/unit/report/report_test.py +# Run the Report Unit Tests +python3 -u $PWD/testing/unit/report/report_test.py + +# Run the RiskProfile Unit Tests +python3 -u $PWD/testing/unit/risk_profile/risk_profile_test.py popd >/dev/null 2>&1 \ No newline at end of file diff --git a/testing/unit/services/output/nmap_report_all_closed.html b/testing/unit/services/output/nmap_report_all_closed.html new file mode 100644 index 000000000..388c673bf --- /dev/null +++ b/testing/unit/services/output/nmap_report_all_closed.html @@ -0,0 +1,483 @@ + + + + + + + + Testrun Report + + + + +

Services Module

+ + + + + + + + + + + + + + +
TCP ports openUDP ports openTotal ports open
000
+ +
+
+ No open ports detected +
+ + + + + + + + Testrun Report + + + + +

Services Module

+ + + + + + + + + + + + + + +
TCP ports openUDP ports openTotal ports open
303
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
PortStateServiceVersion
22/tcpopenssh8.8 protocol 2.0
443/tcpopenhttp
502/tcpopenmbap
+ + + Services Module + + + + + + + + + + + + + + +
TCP ports openUDP ports openTotal ports open
000
+ +
+
+ No open ports detected +
\ No newline at end of file diff --git a/testing/unit/nmap/results/all_closed_scan_result.json b/testing/unit/services/output/services_scan_results.json similarity index 100% rename from testing/unit/nmap/results/all_closed_scan_result.json rename to testing/unit/services/output/services_scan_results.json diff --git a/testing/unit/nmap/reports/nmap_report_all_closed_local.html b/testing/unit/services/reports/nmap_report_all_closed_local.html similarity index 100% rename from testing/unit/nmap/reports/nmap_report_all_closed_local.html rename to testing/unit/services/reports/nmap_report_all_closed_local.html diff --git a/testing/unit/nmap/reports/nmap_report_local.html b/testing/unit/services/reports/nmap_report_local.html similarity index 100% rename from testing/unit/nmap/reports/nmap_report_local.html rename to testing/unit/services/reports/nmap_report_local.html diff --git a/testing/unit/services/results/all_closed_scan_result.json b/testing/unit/services/results/all_closed_scan_result.json new file mode 100644 index 000000000..7d841e869 --- /dev/null +++ b/testing/unit/services/results/all_closed_scan_result.json @@ -0,0 +1,135 @@ +{ + "22tcp": { + "number": "22", + "tcp_udp": "tcp", + "state": "closed", + "service": "ssh", + "version": "8.8 protocol 2.0" + }, + "443tcp": { + "number": "443", + "tcp_udp": "tcp", + "state": "closed", + "service": "http", + "version": "" + }, + "502tcp": { + "number": "502", + "tcp_udp": "tcp", + "state": "closed", + "service": "mbap", + "version": "" + }, + "20udp": { + "number": "20", + "tcp_udp": "udp", + "state": "closed", + "service": "ftp-data", + "version": "" + }, + "21udp": { + "number": "21", + "tcp_udp": "udp", + "state": "closed", + "service": "ftp", + "version": "" + }, + "23udp": { + "number": "23", + "tcp_udp": "udp", + "state": "closed", + "service": "telnet", + "version": "" + }, + "69udp": { + "number": "69", + "tcp_udp": "udp", + "state": "closed", + "service": "tftp", + "version": "" + }, + "80udp": { + "number": "80", + "tcp_udp": "udp", + "state": "closed", + "service": "http", + "version": "" + }, + "109udp": { + "number": "109", + "tcp_udp": "udp", + "state": "closed", + "service": "pop2", + "version": "" + }, + "110udp": { + "number": "110", + "tcp_udp": "udp", + "state": "closed", + "service": "pop3", + "version": "" + }, + "123udp": { + "number": "123", + "tcp_udp": "udp", + "state": "closed", + "service": "ntp", + "version": "" + }, + "143udp": { + "number": "143", + "tcp_udp": "udp", + "state": "closed", + "service": "imap", + "version": "" + }, + "161udp": { + "number": "161", + "tcp_udp": "udp", + "state": "closed", + "service": "snmp", + "version": "" + }, + "220udp": { + "number": "220", + "tcp_udp": "udp", + "state": "closed", + "service": "imap3", + "version": "" + }, + "443udp": { + "number": "443", + "tcp_udp": "udp", + "state": "closed", + "service": "https", + "version": "" + }, + "585udp": { + "number": "585", + "tcp_udp": "udp", + "state": "closed", + "service": "imap4-ssl", + "version": "" + }, + "993udp": { + "number": "993", + "tcp_udp": "udp", + "state": "closed", + "service": "imaps", + "version": "" + }, + "995udp": { + "number": "995", + "tcp_udp": "udp", + "state": "closed", + "service": "pop3s", + "version": "" + }, + "3713udp": { + "number": "3713", + "tcp_udp": "udp", + "state": "closed", + "service": "tftps", + "version": "" + } +} \ No newline at end of file diff --git a/testing/unit/nmap/results/ports_open_scan_result.json b/testing/unit/services/results/ports_open_scan_result.json similarity index 100% rename from testing/unit/nmap/results/ports_open_scan_result.json rename to testing/unit/services/results/ports_open_scan_result.json diff --git a/testing/unit/nmap/nmap_module_test.py b/testing/unit/services/services_module_test.py similarity index 90% rename from testing/unit/nmap/nmap_module_test.py rename to testing/unit/services/services_module_test.py index a175d60d0..a24a21d50 100644 --- a/testing/unit/nmap/nmap_module_test.py +++ b/testing/unit/services/services_module_test.py @@ -12,13 +12,13 @@ # See the License for the specific language governing permissions and # limitations under the License. """Module run all the DNS related unit tests""" -from nmap_module import NmapModule +from services_module import ServicesModule import unittest import os import shutil from testreport import TestReport -MODULE = 'nmap' +MODULE = 'services' # Define the file paths TEST_FILES_DIR = 'testing/unit/' + MODULE @@ -45,10 +45,11 @@ def nmap_module_ports_open_report_test(self): # Move test scan into expected folder src_scan_results_path = os.path.join(RESULTS_DIR, 'ports_open_scan_result.json') - dst_scan_results_path = os.path.join(OUTPUT_DIR, 'nmap_scan_results.json') + dst_scan_results_path = os.path.join( + OUTPUT_DIR, 'services_scan_results.json') shutil.copy(src_scan_results_path, dst_scan_results_path) - nmap_module = NmapModule(module=MODULE, + nmap_module = ServicesModule(module=MODULE, log_dir=OUTPUT_DIR, conf_file=CONF_FILE, results_dir=OUTPUT_DIR, @@ -77,10 +78,11 @@ def nmap_module_ports_open_report_test(self): def nmap_module_report_all_closed_test(self): src_scan_results_path = os.path.join(RESULTS_DIR, 'all_closed_scan_result.json') - dst_scan_results_path = os.path.join(OUTPUT_DIR, 'nmap_scan_results.json') + dst_scan_results_path = os.path.join( + OUTPUT_DIR, 'services_scan_results.json') shutil.copy(src_scan_results_path, dst_scan_results_path) - nmap_module = NmapModule(module=MODULE, + nmap_module = ServicesModule(module=MODULE, log_dir=OUTPUT_DIR, conf_file=CONF_FILE, results_dir=OUTPUT_DIR, diff --git a/testing/unit/tls/tls_module_test.py b/testing/unit/tls/tls_module_test.py index 5ef1cd841..455b4065d 100644 --- a/testing/unit/tls/tls_module_test.py +++ b/testing/unit/tls/tls_module_test.py @@ -315,73 +315,77 @@ def security_tls_client_allowed_protocols_test(self): print(str(test_results)) self.assertTrue(test_results[0]) - def tls_module_report_test(self): - print('\ntls_module_report_test') - os.environ['DEVICE_MAC'] = '38:d1:35:01:17:fe' - pcap_file = os.path.join(CAPTURES_DIR, 'tls.pcap') - tls = TLSModule(module=MODULE, - log_dir=OUTPUT_DIR, - conf_file=CONF_FILE, - results_dir=OUTPUT_DIR, - startup_capture_file=pcap_file, - monitor_capture_file=pcap_file, - tls_capture_file=pcap_file) - report_out_path = tls.generate_module_report() - - with open(report_out_path, 'r', encoding='utf-8') as file: - report_out = file.read() - - # Read the local good report - with open(LOCAL_REPORT, 'r', encoding='utf-8') as file: - report_local = file.read() - - self.assertEqual(report_out, report_local) - - def tls_module_report_ext_test(self): - print('\ntls_module_report_ext_test') - os.environ['DEVICE_MAC'] = '28:29:86:27:d6:05' - pcap_file = os.path.join(CAPTURES_DIR, 'tls_ext.pcap') - tls = TLSModule(module=MODULE, - log_dir=OUTPUT_DIR, - conf_file=CONF_FILE, - results_dir=OUTPUT_DIR, - startup_capture_file=pcap_file, - monitor_capture_file=pcap_file, - tls_capture_file=pcap_file) - report_out_path = tls.generate_module_report() - - # Read the generated report - with open(report_out_path, 'r', encoding='utf-8') as file: - report_out = file.read() - - # Read the local good report - with open(LOCAL_REPORT_EXT, 'r', encoding='utf-8') as file: - report_local = file.read() - - self.assertEqual(report_out, report_local) - - def tls_module_report_no_cert_test(self): - print('\ntls_module_report_no_cert_test') - os.environ['DEVICE_MAC'] = '' - pcap_file = os.path.join(CAPTURES_DIR, 'tls_ext.pcap') - tls = TLSModule(module=MODULE, - log_dir=OUTPUT_DIR, - conf_file=CONF_FILE, - results_dir=OUTPUT_DIR, - startup_capture_file=pcap_file, - monitor_capture_file=pcap_file, - tls_capture_file=pcap_file) - report_out_path = tls.generate_module_report() - - # Read the generated report - with open(report_out_path, 'r', encoding='utf-8') as file: - report_out = file.read() - - # Read the local good report - with open(LOCAL_REPORT_NO_CERT, 'r', encoding='utf-8') as file: - report_local = file.read() - - self.assertEqual(report_out, report_local) + # Commented out whilst TLS report is recreated + # def tls_module_report_test(self): + # print('\ntls_module_report_test') + # os.environ['DEVICE_MAC'] = '38:d1:35:01:17:fe' + # pcap_file = os.path.join(CAPTURES_DIR, 'tls.pcap') + # tls = TLSModule(module=MODULE, + # log_dir=OUTPUT_DIR, + # conf_file=CONF_FILE, + # results_dir=OUTPUT_DIR, + # startup_capture_file=pcap_file, + # monitor_capture_file=pcap_file, + # tls_capture_file=pcap_file) + # report_out_path = tls.generate_module_report() + + # with open(report_out_path, 'r', encoding='utf-8') as file: + # report_out = file.read() + + # # Read the local good report + # with open(LOCAL_REPORT, 'r', encoding='utf-8') as file: + # report_local = file.read() + + # self.assertEqual(report_out, report_local) + + # Commented out whilst TLS report is recreated + # def tls_module_report_ext_test(self): + # print('\ntls_module_report_ext_test') + # os.environ['DEVICE_MAC'] = '28:29:86:27:d6:05' + # pcap_file = os.path.join(CAPTURES_DIR, 'tls_ext.pcap') + # tls = TLSModule(module=MODULE, + # log_dir=OUTPUT_DIR, + # conf_file=CONF_FILE, + # results_dir=OUTPUT_DIR, + # startup_capture_file=pcap_file, + # monitor_capture_file=pcap_file, + # tls_capture_file=pcap_file) + # report_out_path = tls.generate_module_report() + + # # Read the generated report + # with open(report_out_path, 'r', encoding='utf-8') as file: + # report_out = file.read() + + # # Read the local good report + # with open(LOCAL_REPORT_EXT, 'r', encoding='utf-8') as file: + # report_local = file.read() + + # self.assertEqual(report_out, report_local) + + # Commented out whilst TLS report is recreated + # def tls_module_report_no_cert_test(self): + # print('\ntls_module_report_no_cert_test') + # os.environ['DEVICE_MAC'] = '' + # pcap_file = os.path.join(CAPTURES_DIR, 'tls_ext.pcap') + # tls = TLSModule(module=MODULE, + # log_dir=OUTPUT_DIR, + # conf_file=CONF_FILE, + # results_dir=OUTPUT_DIR, + # startup_capture_file=pcap_file, + # monitor_capture_file=pcap_file, + # tls_capture_file=pcap_file) + + # report_out_path = tls.generate_module_report() + + # # Read the generated report + # with open(report_out_path, 'r', encoding='utf-8') as file: + # report_out = file.read() + + # # Read the local good report + # with open(LOCAL_REPORT_NO_CERT, 'r', encoding='utf-8') as file: + # report_local = file.read() + + # self.assertEqual(report_out, report_local) def generate_tls_traffic(self, capture_file,