enhance(cve-feed): re-organize code and cut srctl osv blob into cve feed

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`;

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__

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</details>\n\n<!-- generated by srctl "
+_OSV_SELECT_PATTERN = re.compile(
+    rf"{start_pattern_select}(.*?){end_pattern_select}",
+    re.DOTALL
+)
+
+start_pattern_cut = "<details>\n<summary>OSV format</summary>\n\n```json osv\n"
+end_pattern_cut = "\n```\n\n</details>\n\n<!-- generated by srctl .*?-->"
+_OSV_CUT_PATTERN = re.compile(
+    rf"{start_pattern_cut}.*?{end_pattern_cut}",
+    re.DOTALL
+)
+
+start_pattern_cut_generator = "<!-- generated by srctl "
+end_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)