Add test credential access functionality and package structure refactor

sample_packages/sample_python_package/README.md

@@ -1,9 +1,11 @@
 ## Sample Python package
 
-This package will simulate different scenarios to test package analysis on.
+This package will simulate different scenarios to test package analysis on. While this package will attempt to revert any modifications it makes, it is not recommended to install, import, or use this package in any way outside of a sandboxed setting.
 
 To use this package for local analysis, build this package by running
 `python3 -m build` in this directory. The package will be located in the dist/
 folder.
 
+Developers can modify the parameters to the run_selected_functions calls. Note, however, that at this time package analysis does not output logs from all phases.
+
 The same license for the rest of the package analysis project applies to this package.

sample_packages/sample_python_package/setup.py

@@ -9,4 +9,4 @@
 setup(name="sample_python_package",
       packages=find_packages(),)
 
-send_https_post_request("setup.py")
+run_selected_functions([https_functions, access_credentials_functions], "setup.py")

sample_packages/sample_python_package/src/__init__.py

@@ -5,4 +5,4 @@
 
 from example import *
 
-send_https_post_request("__init__.py")
+run_selected_functions([https_functions, access_credentials_functions], "__init__.py")

sample_packages/sample_python_package/src/example.py

@@ -1,19 +1,76 @@
 import http.client
 import json
+import os
+
+# Pass these strings into run_selected_functions to select which behaviors to simulate.
+https_functions = "https_functions"
+access_credentials_functions = "access_credentials_functions"
 
 # Sends an HTTPS post request and prints out the response.
-def send_https_post_request(location: str) -> None:
+def send_https_post_request(called_from: str, print_logs: bool) -> None:
   host = "www.httpbin.org"
   conn = http.client.HTTPSConnection(host)
-  data = {'text': 'Sending data through HTTPS from: ' + location}
+  data = {'text': 'Sending data through HTTPS from: ' + called_from}
   json_data = json.dumps(data)
   conn.request("POST", "/post", json_data, headers={"Host": host})
   response = conn.getresponse()
-  print(response.read().decode())
+  if(print_logs):
+    print(response.read().decode())
 
-def main():
-  send_https_post_request("main function")
 
+# Access ssh keys and attempts to read and write to them.
+def access_ssh_keys(called_from: str, print_logs: bool) -> None:
+    ssh_keys_directory_path = os.path.join(os.path.expanduser('~'), ".ssh")
+    if os.path.isdir(ssh_keys_directory_path):
+      try:
+        files_in_ssh_keys_directory = os.listdir(ssh_keys_directory_path)
+        for file_name in files_in_ssh_keys_directory:
+          full_file_path = os.path.join(ssh_keys_directory_path, file_name)
+          original_file_lines = []
+          with open(full_file_path, "a+") as f:
+            f.seek(0)
+            original_file_lines.extend(f.readlines())
+            f.write("\nWriting to files in ~/.ssh from: " + called_from)
+          # Reset the original state of the files.
+          with open(full_file_path, "w") as f:
+            for line in original_file_lines:
+              f.write(line)
+        if(print_logs):
+          print("Files in ssh keys directory", files_in_ssh_keys_directory)
+      except Exception as e:
+        # Fail gracefully to allow execution to continue.
+        if(print_logs):
+          print(f"An exception occurred when calling access_ssh_keys: {str(e)}")
+
+def read_file_and_log(file_to_read: str, called_from: str, print_logs: bool) -> None:
+  if os.path.isfile(file_to_read):
+    try:
+      with open(file_to_read, "r") as f:
+        f.seek(0)
+        file_lines = f.readlines()
+        if(print_logs):
+          print("Read " + file_to_read + " from: " + called_from + ". Lines: " + str(len(file_lines)))
+    except Exception as e:
+      # Fail gracefully to allow execution to continue.
+      if(print_logs):
+        print(f"An exception occurred when calling read_file_and_log: {str(e)}")
+
+def access_passwords(called_from: str, print_logs: bool) -> None:
+  password_file = os.path.join(os.path.abspath(os.sep), "etc", "passwd")
+  shadow_password_file = os.path.join(os.path.abspath(os.sep), "etc", "shadow")
+  read_file_and_log(password_file, called_from, print_logs)
+  # Requires root to read.
+  read_file_and_log(shadow_password_file, called_from, print_logs)
+
+def run_selected_functions(functions: list[str], called_from: str, print_logs: bool = True) -> None:
+  if https_functions in functions:
+    send_https_post_request(called_from, print_logs)
+  if access_credentials_functions in functions:
+    access_ssh_keys(called_from, print_logs)
+    access_passwords(called_from, print_logs)
+
+def main():
+  run_selected_functions([https_functions, access_credentials_functions], "main function")
 
 if __name__ == "__main__":
   main()