| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| from app import config |
| import dataclasses |
| import datetime |
| from email.header import decode_header |
| from email.utils import getaddresses |
| from enum import Enum, auto |
| import hashlib |
| import json |
| import pathlib |
| from quart import current_app |
| import re |
| |
| @dataclasses.dataclass(frozen=True) |
| class Reporter: |
| name: str |
| email: str |
| |
| @property |
| def display_name(self) -> str: |
| return self.name or self.email or "unknown" |
| |
| @property |
| def tooltip(self) -> str: |
| if self.name and self.email: |
| return f"{self.name} <{self.email}>" |
| return self.email or self.name or "unknown" |
| |
| @property |
| def initials(self) -> str: |
| source = self.name or self.email.split('@', 1)[0] |
| letters = [w[0] for w in re.split(r'\s+', source) if w] |
| if not letters: |
| return '?' |
| return ''.join(letters[:3]).upper() |
| |
| @property |
| def color(self) -> str: |
| digest = hashlib.md5(self.email.lower().encode()).hexdigest() |
| hue = int(digest[:4], 16) % 360 |
| return f"hsl({hue}, 55%, 45%)" |
| |
| |
| @dataclasses.dataclass(frozen=True) |
| class Report: |
| security_team_name: str |
| """internal label the security team assigned to the thread, |
| this is not a secret but not meant for external communication""" |
| cves: list[str] |
| github: (str, str) |
| """If this project tracks security issues in private GitHub issues, the GitHub issue link (title and url)""" |
| jira: str |
| """If this project tracks security issues in private jira issues, the Jira ID""" |
| title: str |
| messageid: str |
| """message_id of the first email in the thread.""" |
| listid: str |
| """list id of the first email in the thread.""" |
| link: str |
| """link to the email archive for project members who may not |
| necessarily be ASF members.""" |
| reporter: Reporter | None |
| |
| state: str |
| |
| subproject: str | None |
| """For PMCs split across subprojects, the subproject this report belongs to.""" |
| |
| timestamp: datetime.datetime |
| |
| @property |
| def date(self) -> datetime.date: |
| return self.timestamp.date() |
| |
| @property |
| def sanitized_title(self) -> str: |
| cleaned = re.sub(r"[^A-Za-z0-9 .\-()]", " ", self.title) |
| cleaned = re.sub(r"\s+", " ", cleaned).strip() |
| return cleaned[:200] |
| |
| @property |
| def asf_member_link(self) -> str: |
| return _ponymail_link(self.messageid, self.listid) |
| |
| def _known_bad_address(time, address): |
| mailtime = datetime.datetime.fromtimestamp(time, tz=datetime.timezone.utc).date() |
| spark_retirement = datetime.date.fromisoformat("2026-02-16") |
| if address == 'security@spark.apache.org' and spark_retirement < mailtime: |
| return True |
| return False |
| |
| def _apache_list_address(email): |
| addresses = list(getaddresses([email['to']])) |
| if 'cc' in email: |
| addresses.extend(getaddresses([email['cc']])) |
| for _, address in addresses: |
| if address == "officesecurity@lists.freedesktop.org": |
| return "security@openoffice.apache.org" |
| if address.endswith('.apache.org') and not _known_bad_address(email['mailtime'], address): |
| return address |
| return None |
| |
| def _ponymail_link(messageid, listid): |
| partly_encoded_messageid = messageid.replace(' ', '+').replace('+', '%2B').replace('=', '%3D').replace('@', '%40') |
| return f"https://lists.apache.org/thread/{partly_encoded_messageid}?<{listid}>" |
| |
| def _project_link(emails): |
| for email in emails[:5]: |
| list_addr = _apache_list_address(email) |
| if list_addr: |
| return _ponymail_link(email['message_id'], list_addr.replace('@', '.')) |
| return _ponymail_link(emails[0]['message_id'], "security.apache.org") |
| |
| def _reporter(email) -> Reporter | None: |
| addresses = [a for a in getaddresses([email.get('from', '')]) if a[1]] |
| if not addresses: |
| return None |
| name, address = addresses[0] |
| if ' via ' in name and address.endswith('apache.org'): |
| reply_to_addresses = [a for a in getaddresses([email.get('reply_to', '')]) if a[1]] |
| if not reply_to_addresses: |
| return None |
| name, address = reply_to_addresses[0] |
| return Reporter(name=name, email=address) |
| |
| def load_pmc_report(pmc: str, path: pathlib.Path) -> Report | None: |
| with open(path) as f: |
| emails = json.loads(f.read()) |
| |
| m = re.match(r"(?:CVE-\S+\s+)*CVE-\S+", path.name) |
| cves = m.group(0).split() if m else [] |
| |
| jira = None |
| if pmc in config.get().pmcs_using_jira: |
| m = re.match(r"\S+ (\d+) .*", path.name) |
| if m: |
| jira = config.get().pmcs_using_jira[pmc] + "-" + m.group(1) |
| |
| github = None |
| if pmc in config.get().pmcs_using_github: |
| for email in emails: |
| m = re.match(r"^\[.*#(\d+)\)$", email['subj']) |
| if m: |
| issue_nr = m.group(1) |
| github = (f"#{issue_nr}", f"https://github.com/{config.get().pmcs_using_github[pmc]}/issues/{issue_nr}") |
| break |
| |
| if cves: |
| state = "confirmed" |
| else: |
| m = re.match(r".*wf (.*).json", path.name) |
| if not m: |
| state = "untriaged" |
| elif m.groups()[0] == "cve-allocation": |
| state = "confirmed" |
| else: |
| state = m.groups()[0] |
| |
| # the subproject is the word after the leading date or CVE(s) |
| m = re.match(r"(?:(?:CVE-\S+\s+)*CVE-\S+|\d{4}-\d{2}-\d{2})\s+(\w+)", path.name) |
| subproject = m.group(1) if m else None |
| |
| if not emails: |
| print(f"Empty label: {path.name}") |
| return None |
| |
| first_email = emails[0] |
| raw_subject = first_email['subj'] |
| try: |
| title = "".join( |
| s.decode(c or "ascii", errors="replace") if isinstance(s, bytes) else s |
| for s, c in decode_header(raw_subject) |
| ) |
| except Exception: |
| title = raw_subject |
| title = title.strip() or "(untitled)" |
| if title.startswith("[SECURITY] "): |
| title = title.removeprefix("[SECURITY] ") |
| elif title.startswith("[Security] "): |
| title = title.removeprefix("[Security] ") |
| |
| apache_list_address = _apache_list_address(first_email) |
| if apache_list_address: |
| listid = apache_list_address.replace('@', '.') |
| else: |
| listid = 'security.apache.org' |
| |
| return Report( |
| path.name, |
| cves, |
| github, |
| jira, |
| title, |
| first_email['message_id'], |
| listid, |
| _project_link(emails), |
| _reporter(first_email), |
| state, |
| subproject, |
| datetime.datetime.fromtimestamp(first_email['mailtime'], tz=datetime.timezone.utc), |
| ) |
| |
| def _load_reports_dir(pmc: str) -> list[Report]: |
| d = config.get().data_dir_path / pmc |
| threads = list(d.glob('**/*.json')) |
| |
| return [ r for r in (load_pmc_report(pmc, t) for t in threads) if r is not None ] |
| |
| async def load_pmc_reports(pmc: str) -> list[Report]: |
| if not re.fullmatch(r"[a-z0-9]+", pmc): |
| raise ValueError(f"invalid PMC name: {pmc!r}") |
| |
| result = _load_reports_dir(pmc) |
| # attic projects no longer have a PMC, so the security team |
| # handles their reports directly |
| if pmc == "security": |
| for attic_pmc in config.get().pmcs_in_attic: |
| if not re.fullmatch(r"[a-z0-9]+", attic_pmc): |
| continue |
| result.extend( |
| dataclasses.replace(r, subproject=attic_pmc) |
| for r in _load_reports_dir(attic_pmc) |
| ) |
| return result |
