Repository navigation
Fetch MSRC CVE Data #1
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: '0 * * * *' # Hourly (at minute 0) | |
| 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 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_articles = [] | |
| seen_kb = set() | |
| for r in v.get('Remediations', []): | |
| kb_num = (r.get('Description') or {}).get('Value', '').strip() | |
| if kb_num.upper().startswith('KB'): | |
| kb_num = kb_num[2:] | |
| url_str = r.get('URL', '').strip() | |
| fixed = r.get('FixedBuild', '').strip() | |
| subtype = r.get('SubType', '').strip() | |
| pids = r.get('ProductID', []) | |
| if not kb_num or not url_str or 'catalog' not in url_str.lower(): | |
| continue | |
| if kb_num in seen_kb: | |
| for existing in kb_articles: | |
| if existing['kb'] == kb_num: | |
| for pid in pids: | |
| pname = product_map.get(pid, '') | |
| if pname and pname not in existing['products']: | |
| existing['products'].append(pname) | |
| continue | |
| seen_kb.add(kb_num) | |
| kb_products = [product_map.get(pid,'') for pid in pids if product_map.get(pid,'')] | |
| kb_articles.append({ | |
| 'kb': kb_num, | |
| 'url': url_str, | |
| 'fixed_build': fixed, | |
| 'subtype': subtype, | |
| 'products': kb_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 |