Enrique Sánchez has proposed merging ~enriqueesanchz/launchpad:fix-update-cve
into launchpad:master.
Commit message:
Fix `cveimport.py` to support cvelist releases page
Change the midnight cvelist url as it uses the date from the day before.
Add retries to the delta cvelist download as we don't know at which time will
they upload the file.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~enriqueesanchz/launchpad/+git/launchpad/+merge/490181
--
Your team Launchpad code reviewers is requested to review the proposed merge of
~enriqueesanchz/launchpad:fix-update-cve into launchpad:master.
diff --git a/lib/lp/bugs/scripts/cveimport.py b/lib/lp/bugs/scripts/cveimport.py
index d216356..be7d6b2 100644
--- a/lib/lp/bugs/scripts/cveimport.py
+++ b/lib/lp/bugs/scripts/cveimport.py
@@ -13,6 +13,7 @@ import zipfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import urljoin
+from typing import Callable
import defusedxml.ElementTree as ElementTree
import requests
@@ -25,6 +26,7 @@ from zope.lifecycleevent import ObjectModifiedEvent
from lp.bugs.interfaces.cve import CveStatus, ICveSet
from lp.services.config import config
from lp.services.looptuner import ITunableLoop, LoopTuner
+from lp.services.log.logger import LaunchpadLogger
from lp.services.scripts.base import (
LaunchpadCronScript,
LaunchpadScriptFailure,
@@ -149,6 +151,34 @@ def update_one_cve(cve_node, log):
return
+def retry_on_failure(
+ func: Callable, max_retries: int, delay: int, logger: LaunchpadLogger
+):
+ """
+ Retry a function on failure with a fixed delay between retries.
+
+ :param func: The function to call.
+ :param max_retries: Maximum number of retries.
+ :param delay: The fixed delay (in seconds) between retries.
+ :param logger: LaunchpadLogger used to print retry messages.
+ :return: The result of the function if successful.
+ """
+ attempt = 0
+ while attempt < max_retries:
+ try:
+ return func()
+ except LaunchpadScriptFailure as e:
+ attempt += 1
+ if attempt < max_retries:
+ logger.info(
+ f"Retrying... ({attempt}/{max_retries}) - "
+ f"waiting for {delay} seconds..."
+ )
+ time.sleep(delay)
+ else:
+ raise e
+
+
@implementer(ITunableLoop)
class CveUpdaterTunableLoop:
"""An `ITunableLoop` for updating CVEs."""
@@ -193,6 +223,8 @@ class CveUpdaterTunableLoop:
class CVEUpdater(LaunchpadCronScript):
+ max_retries = 6
+
def add_my_options(self):
"""Parse command line arguments."""
self.parser.add_option(
@@ -242,6 +274,10 @@ class CVEUpdater(LaunchpadCronScript):
date_str = f"{year}-{date_str}"
hour = now.hour
+ # Go back 24 hours to get yesterday's date
+ yesterday = now - timedelta(days=1)
+ yesterday_str = yesterday.strftime("%Y-%m-%d")
+
base_url = config.cveupdater.github_cve_url
if delta:
@@ -250,11 +286,8 @@ class CVEUpdater(LaunchpadCronScript):
# A "real" delta update at midnight is empty since we just formed
# a new baseline for the day.
if hour == 0:
- # Go back 24 hours to get yesterday's date
- yesterday = now - timedelta(days=1)
- date_str = yesterday.strftime("%Y-%m-%d")
- release_tag = f"cve_{date_str}_at_end_of_day"
- filename = f"{date_str}_delta_CVEs_at_end_of_day.zip"
+ release_tag = f"cve_{yesterday_str}_at_end_of_day"
+ filename = f"{yesterday_str}_delta_CVEs_at_end_of_day.zip"
else:
# For all hours, use the standard hourly format
hour_str = f"{hour:02d}00"
@@ -262,7 +295,7 @@ class CVEUpdater(LaunchpadCronScript):
filename = f"{date_str}_delta_CVEs_at_{hour_str}Z.zip"
else:
release_tag = f"cve_{date_str}_0000Z"
- filename = f"{date_str}_all_CVEs_at_midnight.zip.zip"
+ filename = f"{yesterday_str}_all_CVEs_at_midnight.zip.zip"
# Construct the full URL
url = urljoin(base_url, f"{release_tag}/{filename}")
@@ -454,7 +487,14 @@ class CVEUpdater(LaunchpadCronScript):
)
# download the ZIP file
- response = self.fetchCVEURL(url)
+ # We need to retry as we don't know the exact time the github
+ # release will be published
+ response = retry_on_failure(
+ lambda: self.fetchCVEURL(url),
+ max_retries=self.max_retries,
+ delay=10 * 60,
+ logger=self.logger,
+ )
# extract to temporary directory
temp_dir = self.extract_github_zip(response, delta=True)
_______________________________________________
Mailing list: https://launchpad.net/~launchpad-reviewers
Post to : launchpad-reviewers@lists.launchpad.net
Unsubscribe : https://launchpad.net/~launchpad-reviewers
More help : https://help.launchpad.net/ListHelp
