Repository navigation
Fetch MSRC CVE Data #37
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| name: Fetch MSRC CVE Data | |
| on: | |
| schedule: | |
| - cron: '*/30 * * * *' # Automatically runs every half hour | |
| workflow_dispatch: # Manual trigger from GitHub UI | |
| jobs: | |
| fetch: | |
| runs-on: ubuntu-latest | |
| permissions: | |
| contents: write | |
| steps: | |
| - name: Checkout | |
| uses: actions/checkout@v4 | |
| - name: Fetch and Parse 12 Months of MSRC Data | |
| run: | | |
| mkdir -p data | |
| python3 << 'EOF' | |
| import json | |
| import re | |
| import time | |
| import urllib.request | |
| import urllib.error | |
| from datetime import datetime, date, timezone | |
| def fetch_with_retry(url, retries=3, delay=2): | |
| req = urllib.request.Request(url, headers={'Accept': 'application/json'}) | |
| for attempt in range(retries): | |
| try: | |
| with urllib.request.urlopen(req, timeout=15) as resp: | |
| return json.loads(resp.read().decode('utf-8')) | |
| except urllib.error.HTTPError as e: | |
| if e.code == 404: | |
| raise e | |
| print(f"-> HTTP Error {e.code} for {url}. Attempt {attempt + 1}/{retries}...") | |
| except Exception as e: | |
| print(f"-> Connection error: {e} for {url}. Attempt {attempt + 1}/{retries}...") | |
| if attempt < retries - 1: | |
| wait_time = delay * (attempt + 1) | |
| print(f"Waiting {wait_time}s before retry...") | |
| time.sleep(wait_time) | |
| raise Exception(f"Failed to fetch {url} after {retries} attempts") | |
| # Generate last 12 months (oldest first, so newest overwrites in case of revisions) | |
| today = date.today() | |
| month_ids = [] | |
| for i in range(11, -1, -1): | |
| m = today.month - i | |
| y = today.year | |
| while m <= 0: | |
| m += 12 | |
| y -= 1 | |
| month_name = date(y, m, 1).strftime('%b') | |
| month_ids.append(f"{y}-{month_name}") | |
| print("Generating data for months:", month_ids) | |
| cves_map = {} | |
| for month_id in month_ids: | |
| url = f"https://api.msrc.microsoft.com/cvrf/v3.0/cvrf/{month_id}" | |
| print(f"Fetching: {url} ...") | |
| try: | |
| raw = fetch_with_retry(url) | |
| except urllib.error.HTTPError as e: | |
| if e.code == 404: | |
| print(f"-> Month {month_id} not published yet (404). Skipping.") | |
| continue | |
| except Exception as e: | |
| print(f"-> Error fetching {month_id}: {e}. Skipping.") | |
| continue | |
| product_map = {} | |
| def walk(branch): | |
| if not branch: | |
| return | |
| for p in branch.get('FullProductName', []): | |
| product_map[p['ProductID']] = p['Value'] | |
| for b in branch.get('Branch', []): | |
| walk(b) | |
| walk(raw.get('ProductTree', {})) | |
| vulns = raw.get('Vulnerability', []) | |
| for v in vulns: | |
| cve_id = v.get('CVE', '') | |
| if not cve_id: | |
| continue | |
| title = '' | |
| if v.get('Title') and v['Title'].get('Value'): | |
| title = v['Title']['Value'] | |
| severity = '' | |
| for t in v.get('Threats', []): | |
| if t.get('Type') == 3: | |
| severity = (t.get('Description') or {}).get('Value', '') | |
| if severity: | |
| break | |
| if not severity: | |
| for t in v.get('Threats', []): | |
| if t.get('Type') == 0: | |
| severity = (t.get('Description') or {}).get('Value', '') | |
| if severity: | |
| break | |
| cvss = None | |
| for s in v.get('CVSSScoreSets', []): | |
| if s.get('BaseScore') and ('3.1' in s.get('Vector','') or '3.0' in s.get('Vector','')): | |
| cvss = float(s['BaseScore']) | |
| break | |
| if cvss is None: | |
| for s in v.get('CVSSScoreSets', []): | |
| if s.get('BaseScore'): | |
| cvss = float(s['BaseScore']) | |
| break | |
| products = [] | |
| for ps in v.get('ProductStatuses', []): | |
| for pid in ps.get('ProductID', []): | |
| pname = product_map.get(pid, '') | |
| if pname and pname not in products: | |
| products.append(pname) | |
| product = products[0] if products else '' | |
| date_str = '' | |
| revs = v.get('RevisionHistory', []) | |
| if revs: | |
| date_str = revs[0].get('Date', '')[:10] | |
| exploited = False | |
| for t in v.get('Threats', []): | |
| if t.get('Type') == 1: | |
| desc = (t.get('Description') or {}).get('Value', '') | |
| if 'exploited:yes' in desc.lower().replace(' ', ''): | |
| exploited = True | |
| break | |
| kb_map = {} | |
| for r in v.get('Remediations', []): | |
| desc_val = (r.get('Description') or {}).get('Value', '').strip() | |
| subtype_val = r.get('SubType', '').strip() | |
| url_str = r.get('URL', '').strip() | |
| fixed = r.get('FixedBuild', '').strip() | |
| pids = r.get('ProductID', []) | |
| # Robustly extract KB number | |
| kb_num = None | |
| if desc_val.upper().startswith('KB') and desc_val[2:].isdigit(): | |
| kb_num = desc_val[2:] | |
| elif desc_val.isdigit() and 6 <= len(desc_val) <= 8: | |
| kb_num = desc_val | |
| elif subtype_val.upper().startswith('KB') and subtype_val[2:].isdigit(): | |
| kb_num = subtype_val[2:] | |
| elif subtype_val.isdigit() and 6 <= len(subtype_val) <= 8: | |
| kb_num = subtype_val | |
| else: | |
| m = re.search(r'\b(?:KB)?(\d{6,8})\b', desc_val, re.IGNORECASE) | |
| if m: | |
| kb_num = m.group(1) | |
| else: | |
| m = re.search(r'\b(?:KB)?(\d{6,8})\b', subtype_val, re.IGNORECASE) | |
| if m: | |
| kb_num = m.group(1) | |
| elif url_str: | |
| m = re.search(r'(?:help/|q=KB|q=)(\d{6,8})\b', url_str, re.IGNORECASE) | |
| if m: | |
| kb_num = m.group(1) | |
| else: | |
| m = re.search(r'\b(\d{6,8})\b', url_str) | |
| if m: | |
| kb_num = m.group(1) | |
| if not kb_num: | |
| continue | |
| kb_products = [product_map.get(pid, '') for pid in pids if product_map.get(pid, '')] | |
| def get_url_score(url): | |
| if not url: return 0 | |
| u_l = url.lower() | |
| if 'catalog.update.microsoft.com' in u_l: return 4 | |
| if 'support.microsoft.com' in u_l: return 3 | |
| if 'microsoft.com' in u_l: return 2 | |
| return 1 | |
| if kb_num not in kb_map: | |
| kb_map[kb_num] = { | |
| 'kb': kb_num, | |
| 'url': url_str, | |
| 'fixed_build': fixed, | |
| 'subtype': subtype_val, | |
| 'products': kb_products | |
| } | |
| else: | |
| # Merge products | |
| for pname in kb_products: | |
| if pname and pname not in kb_map[kb_num]['products']: | |
| kb_map[kb_num]['products'].append(pname) | |
| # Update URL if higher score | |
| if get_url_score(url_str) > get_url_score(kb_map[kb_num]['url']): | |
| kb_map[kb_num]['url'] = url_str | |
| # Update fixed build if empty | |
| if not kb_map[kb_num]['fixed_build'] and fixed: | |
| kb_map[kb_num]['fixed_build'] = fixed | |
| # Update subtype if new is better | |
| if (not kb_map[kb_num]['subtype'] or kb_map[kb_num]['subtype'].isdigit()) and subtype_val and not subtype_val.isdigit(): | |
| kb_map[kb_num]['subtype'] = subtype_val | |
| # Finalize KB map | |
| kb_articles = [] | |
| for kb_num, kb_info in kb_map.items(): | |
| if not kb_info['url']: | |
| kb_info['url'] = f"https://catalog.update.microsoft.com/v7/site/Search.aspx?q=KB{kb_num}" | |
| if not kb_info['subtype'] or kb_info['subtype'].isdigit(): | |
| kb_info['subtype'] = 'Security Update' | |
| kb_articles.append({ | |
| 'kb': kb_info['kb'], | |
| 'url': kb_info['url'], | |
| 'fixed_build': kb_info['fixed_build'], | |
| 'subtype': kb_info['subtype'], | |
| 'products': kb_info['products'][:8] | |
| }) | |
| kb_articles.sort(key=lambda x: x['kb']) | |
| cves_map[cve_id] = { | |
| 'cve_id': cve_id, | |
| 'title': title, | |
| 'severity': severity, | |
| 'cvss': cvss, | |
| 'product': product, | |
| 'products': products[:5], | |
| 'date': date_str, | |
| 'exploited': exploited, | |
| 'kb_articles': kb_articles, | |
| 'kb_count': len(kb_articles), | |
| } | |
| parsed = list(cves_map.values()) | |
| sev_order = {'Critical':0, 'Important':1, 'Moderate':2, 'Low':3} | |
| parsed.sort(key=lambda x: ( | |
| 0 if x['exploited'] else 1, | |
| sev_order.get(x['severity'], 4), | |
| -(x['cvss'] or 0), | |
| x['cve_id'] | |
| )) | |
| now = datetime.now(timezone.utc) | |
| counts = { | |
| 'total': len(parsed), | |
| 'critical': sum(1 for x in parsed if x['severity'] == 'Critical'), | |
| 'important': sum(1 for x in parsed if x['severity'] == 'Important'), | |
| 'moderate': sum(1 for x in parsed if x['severity'] == 'Moderate'), | |
| 'low': sum(1 for x in parsed if x['severity'] == 'Low'), | |
| 'exploited': sum(1 for x in parsed if x['exploited']), | |
| 'patched': sum(1 for x in parsed if x['kb_count'] > 0), | |
| 'updated_at': now.strftime('%Y-%m-%d %H:%M UTC'), | |
| 'month': "Past 12 Months" | |
| } | |
| output = {'meta': counts, 'cves': parsed} | |
| with open('data/msrc_cves.json', 'w', encoding='utf-8') as f: | |
| json.dump(output, f, ensure_ascii=False, indent=2) | |
| print(f"Success: {counts['total']} CVEs consolidated. Critical: {counts['critical']} | Exploited: {counts['exploited']} | Patched: {counts['patched']}") | |
| EOF | |
| - name: Commit updated data | |
| run: | | |
| git config user.name "github-actions[bot]" | |
| git config user.email "github-actions[bot]@users.noreply.github.com" | |
| git add data/msrc_cves.json | |
| git diff --staged --quiet || git commit -m "chore: update rolling 12-month MSRC CVE data [$(date -u +'%Y-%m-%d %H:%M UTC')]" | |
| git push |