Add log level support

Author: enzowritescode

README.md

@@ -8,7 +8,7 @@ Workflow auditing tools to identify security issues in GitHub workflows
 # Usage
 
 ```
-usage: main.py [-h] [--type {repo,org,user}] input
+usage: main.py [-h] [--type {repo,org,user}] [--log-level {debug,info,warning,error,critical}] input
 
 Identify vulnerabilities in GitHub Actions workflow
 
@@ -19,6 +19,8 @@ optional arguments:
   -h, --help            show this help message and exit
   --type {repo,org,user}
                         Type of entity that is being scanned.
+  --log-level {debug,info,warning,error,critical}
+                        Log level for output
 ```
 
 Example:

action_auditor.py

@@ -1,10 +1,6 @@
-from github_wrapper import GHWrapper
-from lib.logger import AuditLogger
 from pathlib import Path
 import re
 
-gh = GHWrapper()
-
 def read_actions_file():
     array_of_usernames = []
     with open('actions.txt','r') as lines:
@@ -16,17 +12,22 @@ def read_actions_file():
                     array_of_usernames.append(username)
     return array_of_usernames
 
-def check_usernames(username_list):
-    for username in username_list:
-        renamed_or_not = gh.stale_checker(username=username)
-        if not renamed_or_not:
-            AuditLogger.warning(f"Security Issue: Supply chain. {username} was renamed but used in workflows. Signup the username at https://github.com to make sure.")
+class ActionAuditor:
+    def __init__(self, gh_wrapper, logger):
+        self.gh = gh_wrapper
+        self.logger = logger
+
+    def check_usernames(self, username_list):
+        for username in username_list:
+            renamed_or_not = self.gh.stale_checker(username=username)
+            if not renamed_or_not:
+                self.logger.warning(f"Security Issue: Supply chain. {username} was renamed but used in workflows. Signup the username at https://github.com to make sure.")
 
-def action_audit():
-    if Path('actions.txt').exists():
-        usernames = read_actions_file()
-        check_usernames(usernames)
-        Path('actions.txt').unlink()
-    else:
-        AuditLogger.info("No actions.txt file to scan. Supply chain scan complete.")
+    def action_audit(self):
+        if Path('actions.txt').exists():
+            usernames = read_actions_file()
+            self.check_usernames(usernames)
+            Path('actions.txt').unlink()
+        else:
+            self.logger.info("No actions.txt file to scan. Supply chain scan complete.")
 

auditor.py

@@ -1,5 +1,4 @@
 from workflow import WorkflowParser, WorkflowVulnAudit
-from lib.logger import AuditLogger
 
 vuln_analyzer = WorkflowVulnAudit()
 
@@ -15,6 +14,7 @@ def risky_trigger_analysis(identified_triggers):
 """
 Input: 
    content - YAML content read from the workflow files.
+   logger - configured logger
 Output:
     scan result (if any) in scan.log file.
 Summary:
@@ -24,7 +24,7 @@ def risky_trigger_analysis(identified_triggers):
     jobs and steps. It then checks the identified key-value pairs
     against known risks through WorkflowParser and WorkflowVulnAudit.
 """
-def content_analyzer(content):
+def content_analyzer(content, logger):
     risky_triggers = []
     all_actions = []
     commands = []
@@ -38,7 +38,7 @@ def content_analyzer(content):
 
         counter = 1 # Counter used to identify which line of code is vulnerable.
         if secrets:
-            AuditLogger.info(f">>> Secrets used in workflow: {','.join(secrets)}")
+            logger.info(f">>> Secrets used in workflow: {','.join(secrets)}")
 
         # Retrieve and store all needed information for a workflow run for analysis.
         if all_jobs:
@@ -49,7 +49,7 @@ def content_analyzer(content):
                 try:
                     environs.update(all_jobs[job].get('env',{}))
                 except:
-                    AuditLogger.info(">> Environ variable is malformed")
+                    logger.info(">> Environ variable is malformed")
                 for step_number,step in enumerate(steps):
                     actions, run_command, with_input, step_environ = workflow_client.analyze_step(step)
                     if actions:
@@ -82,9 +82,9 @@ def content_analyzer(content):
                                         if environ_var_value:
                                             risky_env = vuln_analyzer.risky_command(command_string=environ_var_value)
                                             if risky_env and list(risky_env.keys())[0] != 'environ_regex':
-                                                AuditLogger.warning(f">>> Security Issue: RCE detected with {regex} in {step_number}: ENV variable {environ_variable} is called through GitHub context and takes user input {environ_var_value}")
+                                                logger.warning(f">>> Security Issue: RCE detected with {regex} in {step_number}: ENV variable {environ_variable} is called through GitHub context and takes user input {environ_var_value}")
                                 else:
-                                    AuditLogger.warning(f">>> Security Issue: RCE detected with {regex} in {step_number}: Usage of {','.join(matched_strings)} found.")
+                                    logger.warning(f">>> Security Issue: RCE detected with {regex} in {step_number}: Usage of {','.join(matched_strings)} found.")
 
                 # Some actions combined with triggers can be bad. Check for those cases.
                 action_storage = open('actions.txt','a+')
@@ -100,7 +100,7 @@ def content_analyzer(content):
                                     risky_commits = vuln_analyzer.risky_commit(referenced=ref_value)
                                     if risky_commits:
                                         if 'pull_request_target' in risky_triggers:
-                                            AuditLogger.warning(f">>> Security Issue: Malicious pull request used in actions/checkout. Vulnerable step: {step_number} ")
+                                            logger.warning(f">>> Security Issue: Malicious pull request used in actions/checkout. Vulnerable step: {step_number} ")
                 action_storage.close()
             except Exception as workflow_err:
-                AuditLogger.info(f">>> Error parsing workflow. Error is {str(workflow_err)}")
+                logger.info(f">>> Error parsing workflow. Error is {str(workflow_err)}")

github_wrapper.py

@@ -1,27 +1,28 @@
 import os
 import sys
 import requests
-from lib.logger import AuditLogger
 
 from query_data import return_query, validation_query
 
 """
 Input:
     token - GitHub PAT. Retrieved from environment variable.
+    logger - Configured logger
 
 Summary:
     This wrapper uses GitHub's GraphQL API and repository(ies)
     for the provided scan target. In addition, it is also used
     at the end of the workflow for stale account checks.
 """
 class GHWrapper():
-    def __init__(self):
+    def __init__(self, logger):
         self.token = os.environ.get('PAT',None)
+        self.logger = logger
         if self.token is None:
-            AuditLogger.warning("No GitHub token provided in the PAT env variable. Exiting.")
+            self.logger.warning("No GitHub token provided in the PAT env variable. Exiting.")
             sys.exit()
         if not self.validate_token():
-            AuditLogger.warning("GitHub token provided in the PAT env variable is invalid. Exiting.")
+            self.logger.warning("GitHub token provided in the PAT env variable is invalid. Exiting.")
             sys.exit()
 
     def validate_token(self):
@@ -45,7 +46,7 @@ def call_graphql(self, query):
             return query_request.json()
         else:
             message = query_request.text
-            AuditLogger.error(f"GitHub GraphQL Query failed: {message}")
+            logger.error(f"GitHub GraphQL Query failed: {message}")
             sys.exit(1)
 
     def repo_node_parser(self,repo_node):
@@ -74,11 +75,11 @@ def get_single_repo(self, repo_name):
             if repo_workflows: # this repo has workflows
                 repos_all[repo_name] = repo_workflows
             else:
-                AuditLogger.debug(f"Repo {repo_name} has no workflow.")
+                self.logger.debug(f"Repo {repo_name} has no workflow.")
         return repos_all
 
     def get_multiple_repos(self,target_name,target_type='org'):
-        AuditLogger.info(f"---- Getting repos for {target_name}----")
+        self.logger.info(f"---- Getting repos for {target_name}----")
         repos_all = {}
         query_type = {'org':'organization','user':'user','repo':'repository'}
         try:
@@ -98,16 +99,16 @@ def get_multiple_repos(self,target_name,target_type='org'):
                             repos_all[repo_name] = repo_workflows
                             count += 1
                         else:
-                            AuditLogger.debug(f"Repo {repo_name} has no workflow.")
+                            self.logger.debug(f"Repo {repo_name} has no workflow.")
                     has_more = repos['data'][query_type[target_type]]['repositories']['pageInfo']['hasNextPage']
                     next_cursor = repos['data'][query_type[target_type]]['repositories']['pageInfo']['endCursor']
                     if has_more:
-                        AuditLogger.info("> Retrieve next batch of 100 repos.")
+                        logger.info("> Retrieve next batch of 100 repos.")
                 else:
-                    AuditLogger.error(f"GraphQL response had error.")
+                    self.logger.error(f"GraphQL response had error.")
                     sys.exit(1)
         except Exception as repo_err:
-            AuditLogger.debug(f"Error parsing data. Message: {str(repo_err)}")
+            self.logger.debug(f"Error parsing data. Message: {str(repo_err)}")
         return count, repos_all
 
     def stale_checker(self,username):

lib/logger.py

@@ -1,18 +1,33 @@
 import logging
 
-def build_logger():
+def parse_log_level_input(input):
+    if input == 'debug':
+        level = logging.DEBUG
+    elif input == 'info':
+        level = logging.INFO
+    elif input == 'warning':
+        level = logging.WARNING
+    elif input == 'error':
+        level = logging.ERROR
+    elif input == 'critical':
+        level = logging.CRITICAL
+    else:
+        input = logging.INFO
+
+    return level
+
+def build_logger(log_level='info'):
     log_format = logging.Formatter('%(levelname)s: %(message)s')
     logger = logging.getLogger('Audit Log')
-    logger.setLevel(logging.INFO)
+    log_level = parse_log_level_input(log_level)
+    logger.setLevel(log_level)
 
     channel = logging.StreamHandler()
     channel.setFormatter(log_format)
 
     log_file = logging.FileHandler('scan.log')
-    log_file.setLevel(logging.INFO)
 
     logger.addHandler(channel)
     logger.addHandler(log_file)
     return logger
 
-AuditLogger = build_logger()

main.py

@@ -2,9 +2,9 @@
 
 # Local imports
 from auditor import content_analyzer
-from action_auditor import action_audit
+from action_auditor import ActionAuditor
 from github_wrapper import GHWrapper
-from lib.logger import AuditLogger
+from lib.logger import build_logger
 
 
 """
@@ -18,39 +18,43 @@
     function will call content_analyzer to audit the workflow
     for any potential vulnerabilities. 
 """
-def repo_analysis(repo_workflow):
+def repo_analysis(repo_workflow, logger):
     for workflow in repo_workflow:
         workflow_name = workflow['name']
         workflow_content = workflow['content']
-        AuditLogger.info(f">> Scanning: {workflow_name}")
-        content_analyzer(content=workflow_content) # will print out security issues
+        logger.info(f">> Scanning: {workflow_name}")
+        content_analyzer(content=workflow_content, logger=logger) # will print out security issues
 
 def main():
-    # Supporting user provided arguments: type, and scan target.
+    # Supporting user provided arguments: type, log level, and scan target.
     parser = argparse.ArgumentParser(description='Identify vulnerabilities in GitHub Actions workflow')
     parser.add_argument('--type',choices=['repo','org','user'],
                         help='Type of entity that is being scanned.')
+    parser.add_argument('--log-level',choices=['debug','info','warning','error','critical'], default='info')
     parser.add_argument('input',help='Org, user or repo name (owner/name)')
     args = parser.parse_args()
 
-    gh = GHWrapper()
-    
     target_type = args.type #repo, org, or user
     target_input = args.input #can be repo url, or a username for org/user
+    log_level = args.log_level
+
+    logger = build_logger(log_level)
+    gh = GHWrapper(logger)
 
     if target_type == 'repo':
         repos = gh.get_single_repo(repo_name=target_input)
     else:
         count, repos = gh.get_multiple_repos(target_name=target_input,
                                     target_type=target_type)
-        AuditLogger.info(f"Metric: Scanning total {count} repos")
+        logger.info(f"Metric: Scanning total {count} repos")
 
     for repo_dict in repos:
-        AuditLogger.info(f"> Starting audit of {repo_dict}")
+        logger.info(f"> Starting audit of {repo_dict}")
         repo_workflows = repos[repo_dict]
-        repo_analysis(repo_workflows)
+        repo_analysis(repo_workflows, logger)
 
-    AuditLogger.info(f"> Checking for supply chain attacks.")
-    action_audit()
+    logger.info(f"> Checking for supply chain attacks.")
+    action_auditor = ActionAuditor(gh, logger)
+    action_auditor.action_audit()
 
 main()