DEVOPS-68 automatically create jira issues based on dependabot alerts

Author: githubofkrishnadhas

common_utils.py

@@ -0,0 +1,57 @@
+def limit_string_length(input_string: str, max_length=254):
+    """
+    This function is created to trim the long summary and descriptions of github vumnerability.
+    Jira is having 255 character limits for summaries. so doing for both
+    :param input_string:
+    :param max_length:
+    :return:
+    """
+    # Check the length of the string
+    string_length = len(input_string)
+    print(f"Original string length: {string_length}")
+
+    # Limit the string to max_length characters
+    if string_length > max_length:
+        input_string = input_string[:max_length]
+        print(f"String was truncated to {max_length} characters.")
+
+    return input_string
+
+
+def compare_summaries_from_epic_and_scan(epic_details: list[dict], issues_list:list[dict]):
+    """
+    this function is to compare the stories existing in epic with the github dependabot alerts
+    available to not create duplicate stories. this works to ensure that duplicate wont happen
+    -POC-
+    :param epic_details:
+    :param issues_list:
+    :return:
+    """
+    create_story_true = []
+    print(epic_details)
+    print(issues_list)
+    for issue in issues_list:
+        summary_from_gh = f"{issue['cve_id']} - {issue['repository']} - {issue['issue_severity']}"
+        summary_from_gh_split = summary_from_gh.split(' - ')
+        found_matching_story = False
+        for story in epic_details:
+            story_summary = story['issue_summary']
+            story_summary_split = story_summary.split(' - ')
+            if summary_from_gh_split[:3] == story_summary_split[:3]:
+                print(f'Story already created for this. Story key {story["key"]}')
+                found_matching_story = True
+                break
+        if not found_matching_story:
+            create_story_true.append(issue)
+        # create_story_true.append(issue)
+    return create_story_true
+
+
+
+def main():
+    """ To test the code"""
+    original_string = "This is a very long string..."  # Replace with your actual string
+    limited_string = limit_string_length(original_string)
+
+if __name__ == "__main__":
+    main()

date_time.py

@@ -5,7 +5,6 @@ def date_time():
     now = datetime.now()
     formatted_datetime = now.strftime("%d-%B-%Y %H:%M")
     return formatted_datetime
-    # print(formatted_datetime)
 
 def main():
     """ test code """

github_scanning.py

@@ -1,6 +1,6 @@
 import os
 from dotenv import load_dotenv
-from jira import create_story_under_epic
+# from jira import create_story_under_epic
 from date_time import date_time
 import requests
 
@@ -17,21 +17,22 @@ def scan_for_dependabot_alerts_and_issues(organization: str, repository: str):
     }
     response = requests.get(url=api_endpoint, headers=headers)
     response_json = response.json()
-    print(response_json)
+    # print(response_json)
     issues_list = []
     for issue in response_json:
         dict = {}
         dict['dependancy'] = issue['dependency']['package']
         dict['issue_is_at'] = issue['dependency']['manifest_path']
         dict['cve_id'] = issue['security_advisory']['cve_id']
-        dict['issue_summary'] = issue['security_advisory']['summary']
+        dict['issue_summary'] = f'{issue["security_advisory"]["summary"]} - {repository}'
         dict['issue_description'] = issue['security_advisory']['description']
         dict['issue_severity'] = issue['security_advisory']['severity']
         dict['issue_created_at'] = issue['created_at']
+        dict['repository'] = repository
+
 
         # add dict to list
         issues_list.append(dict)
-        print(issues_list)
 
     return issues_list
 
@@ -42,11 +43,6 @@ def main():
     repository = 'programatically-create-delete-update-github-repository-secrets'
 
     issues_list = scan_for_dependabot_alerts_and_issues(organization=organization, repository=repository)
-    for issues in issues_list:
-        summary = f"{issues['cve_id']} {repository} - {issues['issue_severity']} - {issues['issue_summary']}"
-        description = f"{issues['issue_description']} \n 'Issue Created at' - {issues['issue_created_at']}"
-        story = create_story_under_epic(epic_key='DEVOPS-74', summary=summary, description=description)
-        print(f"Created story {story['key']} at {date_time()} IST")
 
 
 if __name__ == "__main__":

jira.py

@@ -1,19 +1,33 @@
 import os
-import json
+import argparse
 import requests
+from requests.auth import HTTPBasicAuth
 from dotenv import load_dotenv
 from atlassian import Jira
-
-# Initializing jira client
-load_dotenv()
-JIRA_URL = os.getenv('JIRA_URL')
-JIRA_USERNAME = os.getenv('JIRA_USERNAME')
-JIRA_PASSWORD = os.getenv('JIRA_PASSWORD')
-
+from github_scanning import scan_for_dependabot_alerts_and_issues
+from repos import list_repos_in_org
+from date_time import date_time
+from common_utils import limit_string_length, compare_summaries_from_epic_and_scan
 
 
 # Function to create a story under a specific epic
-def create_story_under_epic(epic_key, summary, description):
+def create_story_under_epic(epic_key: str, summary: str, description: str):
+    """
+    Create jira task under a specific epic
+    :param epic_key:
+    :param summary:
+    :param description:
+    :return:
+    """
+    # Initializing jira client
+    load_dotenv()
+    JIRA_URL = os.getenv('JIRA_URL')
+    JIRA_USERNAME = os.getenv('JIRA_USERNAME')
+    JIRA_PASSWORD = os.getenv('JIRA_PASSWORD')
+    jira = Jira(url=os.getenv('JIRA_URL'), username=os.getenv('JIRA_USERNAME'), password=os.getenv('JIRA_PASSWORD'),
+                cloud=True)
+    # for issue in unmatched_issues:
+    #     print('h')
     issue_dict = {
         'project': {'key': 'DEVOPS'},
         'summary': summary,
@@ -24,48 +38,110 @@ def create_story_under_epic(epic_key, summary, description):
         }
     }
     new_issue = jira.issue_create(fields=issue_dict)
-    # pic = new_issue.update(fields={'parent': {'id': epic_key}})
+    print(f"Created story {new_issue['key']} with epic {epic_key} at {date_time()} IST")
 
-    return new_issue
 
-
-def issues_in_epic(epic_key: str, board_id: int):
+def issues_in_epic(epic_key: str):
     """
     list available issues in a epic
     :param epic_key:
     :return:
     """
+    # Initializing jira client
+    load_dotenv()
+    JIRA_URL = os.getenv('JIRA_URL')
+    JIRA_USERNAME = os.getenv('JIRA_USERNAME')
+    JIRA_PASSWORD = os.getenv('JIRA_PASSWORD')
     jira = Jira(url=os.getenv('JIRA_URL'), username=os.getenv('JIRA_USERNAME'), password=os.getenv('JIRA_PASSWORD'),
                 cloud=True)
 
-    # todo
+    url = f'{JIRA_URL}/rest/api/3/search'
+
+    auth = HTTPBasicAuth(JIRA_USERNAME, JIRA_PASSWORD)
+
+    headers = {
+        "Accept": "application/json"
+    }
+
+    query = {
+        'jql': f'Parent = {epic_key}'
+    }
+    response = requests.request(
+        "GET",
+        url,
+        headers=headers,
+        params=query,
+        auth=auth
+    )
+    response_json = response.json()
+    response_json_issues = response_json['issues']
+    if not response_json_issues:
+        issues_in_epic = []
+    else:
+        issues_in_epic = response_json['issues']
+
+    epic_details = []
+    for issues in issues_in_epic:
+        issue = {}
+        issue['key'] = issues['key']
+        issue['issue_type'] = issues['fields']['issuetype']['name']
+        issue['issue_summary'] = issues['fields']['summary']
+        issue['description'] = 'NA'
+        epic_details.append(issue)
+
+    return epic_details
+
+
+def compare_summaries_and_create_story(repo_names: list, organization: str, epic_key: str):
+    """
+    The function helps create stories by calling another function.
+    :param issues_list:
+    :param epic_details:
+    :return:
+    """
+    epic_details = issues_in_epic(epic_key=epic_key)
+
+    for repo in repo_names:
+        # scan details
+        issues_list = scan_for_dependabot_alerts_and_issues(organization=organization, repository=repo)
+        if not issues_list:
+            print(f'No dependabot security alerts found on {organization}/{repo} repository')
+            continue
+        create_story_true = compare_summaries_from_epic_and_scan(issues_list=issues_list, epic_details=epic_details)
+        for issue in create_story_true:
+            summary_from_gh = f"{issue['cve_id']} - {issue['repository']} - {issue['issue_severity']} - {issue['issue_summary']}"
+            description_from_gh = issue['issue_description']
+            summary_after_character_length_check = limit_string_length(input_string=summary_from_gh)
+            description_after_character_length_check = limit_string_length(input_string=description_from_gh)
+
+            create_story_under_epic(summary=summary_after_character_length_check,
+                                    description=description_after_character_length_check, epic_key=epic_key)
+
 
-def get_board_id(project_key: str):
-    # jira = Jira(url=os.getenv('JIRA_URL_0'), username=os.getenv('JIRA_USERNAME'), password=os.getenv('JIRA_PASSWORD'),
-    #             cloud=True)
-    response = requests.get(url=os.getenv('JIRA_URL_0'),auth=(os.getenv('JIRA_USERNAME'), os.getenv('JIRA_PASSWORD')))
-    if response.status_code == 200:
-        boards = response.json()['values']
-        if boards:
-            # Assuming you want the first board ID
-            return boards[0]['id']
-    return None
 
 def main():
     """testing script"""
-    # Replace these values with your specific details
-    epic_key = 'DEVOPS-74'
-    summary = 'New Story Summary135'
-    description = 'Description -- of the new story'
 
-    # Create the story
-    # story = create_story_under_epic(epic_key, summary, description)
-    # print(f"Created story {story['key']} under epic {epic_key}")
+    parser = argparse.ArgumentParser(description='Create Jira issues prograatically based on dependabot alerts')
+    parser.add_argument("--epic_key", help="jira epic ID", type=str, required=True)
+    parser.add_argument('--organization', help='Github org name', type=str, required=True)
+
+    args = parser.parse_args()
+
+    epic_key = args.epic_key
+    organization = args.organization
 
     # List stories under an epic
-    stories = issues_in_epic(epic_key=epic_key, board_id=1)
+    epic_details = issues_in_epic(epic_key=epic_key)
+
+    # Repos in GitHub org
+    repo_names = list_repos_in_org(org_name=organization)
+
+    # compare summaries to decide to create a new story or not
+    compare_summaries_and_create_story(repo_names=repo_names, organization=organization,
+                                       epic_key=epic_key)
+
 
-    # get_board_id('DEVOPS')
 
 if __name__ == "__main__":
     main()

repos.py

@@ -5,7 +5,7 @@
 
 def list_repos_in_org(org_name: str):
     """
-    list all repos in github organization
+    list all repos in github organization using rest api repo end point
     :param org_name:
     :return:
     """