diff --git a/sig-security-tooling/cve-feed/README.md b/sig-security-tooling/cve-feed/README.md index 99a90f8..1e51170 100644 --- a/sig-security-tooling/cve-feed/README.md +++ b/sig-security-tooling/cve-feed/README.md @@ -53,6 +53,26 @@ A script in the [kubernetes/sig-security](https://github.com/kubernetes/sig-secu repository under the [sig-security-tooling/cve-feed/hack](https://github.com/kubernetes/sig-security/tree/main/sig-security-tooling/cve-feed/hack) folder is responsible for generating and updating the feed. +#### Local development + +To run the feed scripts locally you need Python 3 and pip3. Install dependencies from the `hack` directory: + +```bash +cd sig-security-tooling/cve-feed/hack +pip3 install -r requirements.txt +``` + +If your system restricts global package installs (e.g. externally managed environment), use a virtual environment: + +```bash +cd sig-security-tooling/cve-feed +python3 -m venv .venv +source .venv/bin/activate # On Windows: .venv\Scripts\activate +pip3 install -r hack/requirements.txt +``` + +Then run the Python script from the `hack` directory: `python3 fetch-official-cve-feed.py`. + This bash script, named `fetch-cve-feed.sh`: - sets up the Python 3 environment; - generates the CVE feed file with `fetch-official-cve-feed.py`; diff --git a/sig-security-tooling/cve-feed/hack/.gitignore b/sig-security-tooling/cve-feed/hack/.gitignore index 13bf3fc..d3d2971 100644 --- a/sig-security-tooling/cve-feed/hack/.gitignore +++ b/sig-security-tooling/cve-feed/hack/.gitignore @@ -1,3 +1,7 @@ #files generated by cve feed prow job cve-feed-hash official-cve-feed.json + +#file directories generated in python dev +.venv +__pycache__ \ No newline at end of file diff --git a/sig-security-tooling/cve-feed/hack/fetch-official-cve-feed.py b/sig-security-tooling/cve-feed/hack/fetch-official-cve-feed.py index d8faece..74accc1 100755 --- a/sig-security-tooling/cve-feed/hack/fetch-official-cve-feed.py +++ b/sig-security-tooling/cve-feed/hack/fetch-official-cve-feed.py @@ -18,10 +18,79 @@ import json import requests import sys +import re + from datetime import datetime, timezone from cve_title_parser import parse_cve_title -def getCVEStatus(state, state_reason): +# to understand the pattern choices refer to: +# https://github.com/kubernetes/sig-security/blob/main/sig-security-tooling/srctl/state/issue.tmpl + +start_pattern_select = "```json osv\n" +end_pattern_select = "\n```\n\n\n\n" +_OSV_CUT_PATTERN = re.compile( + rf"{start_pattern_cut}.*?{end_pattern_cut}", + re.DOTALL +) + +start_pattern_cut_generator = "" +_OSV_CUT_PATTERN_GENERATOR = re.compile( + rf"{start_pattern_cut_generator}.*?{end_pattern_cut_generator}", + re.DOTALL +) + +# Value for osv_generator when OSV is fetched from cve-feed-osv repo. +OSV_GENERATOR_FROM_FEED_REPO = "OSV from kubernetes-sigs/cve-feed-osv GitHub repository" + + +def gh_issue_osv_extraction(gh_issue_content, cve_in_issue, cve_id, gh_issue_name): + matches = _OSV_SELECT_PATTERN.findall(gh_issue_content) + if not matches or cve_in_issue >= len(matches): + return None + try: + osv_info = json.loads(matches[cve_in_issue].strip()) + return osv_info + except json.JSONDecodeError: + print(f"{cve_id} in issue {gh_issue_name}: invalid osv json") + return None + + +def get_osv_generator(gh_issue_content, cve_in_issue): + matches = _OSV_CUT_PATTERN_GENERATOR.findall(gh_issue_content) + if not matches or cve_in_issue >= len(matches): + return OSV_GENERATOR_FROM_FEED_REPO + osv_generator = matches[cve_in_issue].strip() + return osv_generator + + +def get_osv_info(gh_issue_content, cve_in_issue, cve_id, gh_issue_name): + """Resolve OSV data: try embedded JSON in github issue body first, then fetch from cve-feed-osv repo.""" + osv = gh_issue_osv_extraction(gh_issue_content, cve_in_issue, cve_id, gh_issue_name) + if osv is not None: + return osv + osv_url = f'https://raw.githubusercontent.com/kubernetes-sigs/cve-feed-osv/main/vulns/{cve_id}.json' + try: + res = requests.get(osv_url, timeout=5) + if res.status_code == 200: + return res.json() + except requests.RequestException as e: + print(f"Error fetching OSV for CVE {cve_id}: {e}", file=sys.stderr) + return None + + +def gh_issue_content_without_osv_info(gh_issue_content): + return _OSV_CUT_PATTERN.sub("", gh_issue_content).strip() + + +def get_CVE_status(state, state_reason): if state == "open": if state_reason == "reopened": return "unknown" @@ -33,85 +102,116 @@ def getCVEStatus(state, state_reason): if state_reason == "completed": return "fixed" -url = 'https://api.github.com/search/issues?q=is:issue+label:official-cve-feed+\ -repo:kubernetes/kubernetes&per_page=100' - -headers = {'Accept': 'application/vnd.github.v3+json'} -res = requests.get(url, headers=headers) -gh_items = res.json()['items'] -# Use link header to iterate over pages -# https://docs.github.com/en/rest/overview/resources-in-the-rest-api#pagination -# https://datatracker.ietf.org/doc/html/rfc5988 -# Please note that if there is a great number of pages, this unauthenticated -# request may be subject to rate limits and fail. -# https://docs.github.com/en/rest/overview/resources-in-the-rest-api#rate-limiting -while 'next' in res.links: - res = requests.get(res.links['next']['url'], headers=headers) - gh_items.extend(res.json()['items']) - -feed_envelope = { - 'version': 'https://jsonfeed.org/version/1.1', - 'title': 'Kubernetes Vulnerability Announcements - CVE Feed', - 'home_page_url': 'https://kubernetes.io', - 'feed_url': 'https://kubernetes.io/docs/reference/issues-security/official-cve-feed/index.json', - 'description': 'Auto-refreshing official CVE feed for Kubernetes repository', - 'authors': [ - { - 'name': 'Kubernetes Community', - 'url': 'https://www.kubernetes.dev' - } - ], - '_kubernetes_io': None, - 'items': None, -} -# format the timestamp the same way as GitHub RFC 3339 timestamps, with only seconds and not milli and microseconds. -root_kubernetes_io = {'feed_refresh_job': 'https://testgrid.k8s.io/sig-security-cve-feed#auto-refreshing-official-cve-feed', - 'updated_at': datetime.now(timezone.utc).replace(tzinfo=None).isoformat(sep='T', timespec='seconds') + 'Z'} -feed_envelope['_kubernetes_io'] = root_kubernetes_io - -cve_list = [] -non_parsable_cve_list = [] -for item in gh_items: +def get_gh_cve_issues(): + url = 'https://api.github.com/search/issues?q=is:issue+label:official-cve-feed+\ + repo:kubernetes/kubernetes&per_page=100' + + headers = {'Accept': 'application/vnd.github.v3+json'} + res = requests.get(url, headers=headers) + gh_items = res.json()['items'] + # Use link header to iterate over pages + # https://docs.github.com/en/rest/overview/resources-in-the-rest-api#pagination + # https://datatracker.ietf.org/doc/html/rfc5988 + # Please note that if there is a great number of pages, this unauthenticated + # request may be subject to rate limits and fail. + # https://docs.github.com/en/rest/overview/resources-in-the-rest-api#rate-limiting + while 'next' in res.links: + res = requests.get(res.links['next']['url'], headers=headers) + gh_items.extend(res.json()['items']) + + return gh_items + +def create_json_feed_scaffold(): + feed_envelope = { + 'version': 'https://jsonfeed.org/version/1.1', + 'title': 'Kubernetes Vulnerability Announcements - CVE Feed', + 'home_page_url': 'https://kubernetes.io', + 'feed_url': 'https://kubernetes.io/docs/reference/issues-security/official-cve-feed/index.json', + 'description': 'Auto-refreshing official CVE feed for Kubernetes repository', + 'authors': [ + { + 'name': 'Kubernetes Community', + 'url': 'https://www.kubernetes.dev' + } + ], + '_kubernetes_io': None, + 'items': None, + } + # format the timestamp the same way as GitHub RFC 3339 timestamps, with only seconds and not milli and microseconds. + root_kubernetes_io = {'feed_refresh_job': 'https://testgrid.k8s.io/sig-security-cve-feed#auto-refreshing-official-cve-feed', + 'updated_at': datetime.now(timezone.utc).replace(tzinfo=None).isoformat(sep='T', timespec='seconds') + 'Z'} + feed_envelope['_kubernetes_io'] = root_kubernetes_io + + return feed_envelope + +def create_cve_item_base(gh_issue_item): # These keys respects the item jsonfeed spec https://www.jsonfeed.org/version/1.1/ - cve = {'content_text': None, 'date_published': None, 'external_url': None, + cve_scaffold = {'content_text': None, 'date_published': None, 'external_url': None, 'id': None,'summary': None, 'url': None, '_kubernetes_io': None} # This is a custom extension - item_kubernetes_io = {'google_group_url': None, 'issue_number': None} - cve['_kubernetes_io'] = item_kubernetes_io + item_kubernetes_io = {'google_group_url': None, 'issue_number': None, 'osv': None, 'osv_generator': None} + cve_scaffold['_kubernetes_io'] = item_kubernetes_io - cve['url'] = item['html_url'] - cve['_kubernetes_io']['issue_number'] = item['number'] - cve['content_text'] = item['body'] - cve['date_published'] = item['created_at'] - cve['status'] = getCVEStatus(item['state'], item['state_reason']) + # One GitHub issue can contain multiple CVEs. + # The below items remain the the same for each CVE in a shared GitHub issue. + cve_scaffold['url'] = gh_issue_item['html_url'] + cve_scaffold['_kubernetes_io']['issue_number'] = gh_issue_item['number'] + cve_scaffold['content_text'] = gh_issue_item['body'] + cve_scaffold['date_published'] = gh_issue_item['created_at'] + cve_scaffold['status'] = get_CVE_status(gh_issue_item['state'], gh_issue_item['state_reason']) - try: - cve_ids, description = parse_cve_title(item['title']) - cve['summary'] = description - - first_cve_id = cve_ids[0] - cve['id'] = first_cve_id - cve['external_url'] = f'https://www.cve.org/cverecord?id={first_cve_id}' - cve['_kubernetes_io']['google_group_url'] = f'https://groups.google.com/g/kubernetes-announce/search?q={first_cve_id}' - - # Add additional entries for any remaining CVE IDs - for additional_cve_id in cve_ids[1:]: - additional_cve = copy.deepcopy(cve) - additional_cve['id'] = additional_cve_id - additional_cve['external_url'] = f'https://www.cve.org/cverecord?id={additional_cve_id}' - additional_cve['_kubernetes_io']['google_group_url'] = f'https://groups.google.com/g/kubernetes-announce/search?q={additional_cve_id}' - cve_list.append(additional_cve) - - cve_list.append(cve) - except LookupError: - non_parsable_cve_list.append((item['title'], item['html_url'])) - -feed_envelope['items'] = cve_list -json_feed = json.dumps(feed_envelope, sort_keys=False, indent=4) -print(json_feed) + return cve_scaffold + + +def create_cve_items_list(gh_cve_issues): + cve_list = [] + non_parsable_cve_list = [] + for gh_cve_issue in gh_cve_issues: + cve = create_cve_item_base(gh_cve_issue) + gh_issue_full_body = cve['content_text'] = gh_cve_issue['body'] + cve['content_text'] = gh_issue_content_without_osv_info(gh_cve_issue['body']) + + try: + cve_ids, description = parse_cve_title(gh_cve_issue['title']) + cve['summary'] = description -if len(non_parsable_cve_list) != 0: - print("Failed to parse below CVE issues:", file=sys.stderr) - for title, url in non_parsable_cve_list: - print(f"{title}\n{url}", file=sys.stderr) - exit(7) + cve_count = 0 + first_cve_id = cve_ids[cve_count] + cve['id'] = first_cve_id + cve['external_url'] = f'https://www.cve.org/cverecord?id={first_cve_id}' + cve['_kubernetes_io']['google_group_url'] = f'https://groups.google.com/g/kubernetes-announce/search?q={first_cve_id}' + cve['_kubernetes_io']['osv'] = get_osv_info(gh_issue_full_body, cve_count, first_cve_id, gh_cve_issue['title']) + if cve['_kubernetes_io']['osv'] is not None: + cve['_kubernetes_io']['osv_generator'] = get_osv_generator(gh_issue_full_body, cve_count) + + # Add additional entries for any remaining CVE IDs + for additional_cve_id in cve_ids[1:]: + cve_count = cve_count + 1 + additional_cve = copy.deepcopy(cve) + additional_cve['id'] = additional_cve_id + additional_cve['external_url'] = f'https://www.cve.org/cverecord?id={additional_cve_id}' + additional_cve['_kubernetes_io']['google_group_url'] = f'https://groups.google.com/g/kubernetes-announce/search?q={additional_cve_id}' + additional_cve['_kubernetes_io']['osv'] = get_osv_info(gh_issue_full_body, cve_count, additional_cve_id, gh_cve_issue['title']) + if additional_cve['_kubernetes_io']['osv'] is not None: + additional_cve['_kubernetes_io']['osv_generator'] = get_osv_generator(gh_issue_full_body, cve_count) + cve_list.append(additional_cve) + + cve_list.append(cve) + except LookupError: + non_parsable_cve_list.append((gh_cve_issue['title'], gh_cve_issue['html_url'])) + + if len(non_parsable_cve_list) != 0: + print("Failed to parse below CVE issues:", file=sys.stderr) + for title, url in non_parsable_cve_list: + print(f"{title}\n{url}", file=sys.stderr) + exit(7) + + return cve_list + + +gh_cve_issues = get_gh_cve_issues() +feed_envelope = create_json_feed_scaffold() +feed_envelope['items'] = create_cve_items_list(gh_cve_issues) + +json_feed: str = json.dumps(feed_envelope, sort_keys=False, indent=4) +print(json_feed)