@@ -46,19 +46,28 @@ def steps(cls):
4646 cls .import_new_advisories ,
4747 )
4848
49- # num of advisories
5049 def advisories_count (self ) -> int :
5150 return fetch_count_advisories (self .root_url )
5251
53- # parse the response data
5452 def collect_advisories (self ) -> Iterable [AdvisoryData ]:
5553 raw_data = fetch_advisory_data (self .root_url )
5654 for data in raw_data :
5755 yield to_advisory_data (data )
5856
5957
60- # fetch the html content
6158def fetch_html_response (url ):
59+ """
60+ Fetch and parse the HTML content of a given URL.
61+
62+ This function sends a request to the URL, retrieves the HTML content,
63+ and parses it using BeautifulSoup.
64+
65+ Args:
66+ url (str): The URL to fetch the HTML content from.
67+
68+ Returns:
69+ A BeautifulSoup object representing the parsed HTML content.
70+ """
6271 try :
6372 response = fetch_response (url ).content
6473 soup = BeautifulSoup (response , "html.parser" )
@@ -68,16 +77,72 @@ def fetch_html_response(url):
6877
6978
7079def fetch_count_advisories (url ):
80+ """
81+ Gives the number of advisories from the given URL.
82+ Advisories are identified by <h3> tags.
83+
84+ Args:
85+ url (str): The URL to fetch the advisories from.
86+
87+ Returns:
88+ int: The number of advisories found on the page.
89+
90+ Doctests:
91+ >>> from unittest.mock import patch
92+ >>> from bs4 import BeautifulSoup
93+ >>> from vulnerabilities.pipelines.openssl_importer import fetch_count_advisories
94+ >>> mock_html = '<html><body><h3>Advisory 1</h3><h3>Advisory 2</h3></body></html>'
95+ >>> with patch('vulnerabilities.pipelines.openssl_importer.fetch_html_response') as mock_fetch:
96+ ... mock_fetch.return_value = BeautifulSoup(mock_html, "html.parser")
97+ ... count = fetch_count_advisories("http://example.com")
98+ >>> count
99+ 2
100+ """
101+
71102 soup = fetch_html_response (url )
72103 advisories = soup .find_all ("h3" )
73104 return len (advisories )
74105
75106
76- # fetch the content from the html data
77107def fetch_advisory_data (url ):
108+ """
109+ Fetch advisory data from the given URL.
110+
111+ Args:
112+ url (str): The URL to fetch the advisory data from.
113+
114+ Returns:
115+ list: A list of dictionaries, where each dictionary contains advisory details.
116+
117+ Doctests:
118+ >>> from unittest.mock import patch
119+ >>> from bs4 import BeautifulSoup
120+ >>> from vulnerabilities.pipelines.openssl_importer import fetch_advisory_data
121+ >>> mock_html = '''
122+ ... <html>
123+ ... <body>
124+ ... <h3 id="CVE-2024-12797">
125+ ... <a href="#CVE-2024-12797">CVE-2024-12797</a>
126+ ... </h3>
127+ ... <dl>
128+ ... <dt>Published at</dt>
129+ ... <dd>11 February 2025</dd>
130+ ... </dl>
131+ ... </body>
132+ ... </html>
133+ ... '''
134+ >>> with patch('vulnerabilities.pipelines.openssl_importer.fetch_html_response') as mock_fetch:
135+ ... mock_fetch.return_value = BeautifulSoup(mock_html, "html.parser")
136+ ... advisories = fetch_advisory_data("http://example.com")
137+ >>> len(advisories)
138+ 1
139+ >>> advisories[0]["CVE"]
140+ 'CVE-2024-12797'
141+ """
142+
78143 advisories = []
79144 soup = fetch_html_response (url )
80- # all the CVEs are h3 with id="CVE-.."
145+
81146 for cve_section in soup .find_all ("h3" ):
82147 data_output = {
83148 "date_published" : "" ,
@@ -88,44 +153,33 @@ def fetch_advisory_data(url):
88153 "severity" : "" ,
89154 }
90155
91- # CVE is in a link
92156 data_output ["CVE" ] = cve_section .find ("a" ).text
93157
94- # the <dl> tag in this section
95158 dl = cve_section .find_next_sibling ("dl" )
96- for dt , dd in zip (
97- dl .find_all ("dt" ), dl .find_all ("dd" )
98- ): # combines both the lists,for better iteration
159+ for dt , dd in zip (dl .find_all ("dt" ), dl .find_all ("dd" )):
99160 key = dt .text
100161 value = dd .text
101162
102- # Severity
103163 if key == "Severity" :
104164 data_output ["severity" ] = value
105- # Published Date
106165 elif key == "Published at" :
107166 data_output ["date_published" ] = value
108- # Affected Packages
109167 elif key == "Affected" :
110168 affected_list = [li .text .strip () for li in dd .find_all ("li" )]
111169 data_output ["affected_packages" ] = affected_list
112- # references
113170 elif key == "References" :
114171 references = [a ["href" ] for a in dd .find_all ("a" )]
115172 data_output ["references" ] = references
116173
117- # for summary
118174 for sibling in dl .find_next_siblings ():
119175 if sibling .name == "h2" or sibling .name == "h3" :
120176 break
121177 if sibling .name == "p" :
122178 if "Issue summary:" in sibling .text :
123179 data_output ["summary" ] = sibling .text .strip ("Issue summary:" )
124180
125- # append all the output data to the list
126181 advisories .append (data_output )
127182
128- # return the list with all the advisory data
129183 return advisories
130184
131185
@@ -145,18 +199,48 @@ def fetch_advisory_data(url):
145199"""
146200
147201
148- # parse the advisory data
149202def to_advisory_data (raw_data ) -> AdvisoryData :
150- # alias
203+ """
204+ Convert raw advisory data into an AdvisoryData object.
205+
206+ Args:
207+ raw_data (dict): A dictionary containing raw advisory data.
208+
209+ Returns:
210+ AdvisoryData: An object containing structured advisory information.
211+
212+ Doctests:
213+ >>> from unittest.mock import patch
214+ >>> from datetime import datetime, timezone
215+ >>> from vulnerabilities.pipelines.openssl_importer import to_advisory_data
216+ >>> raw_data = {
217+ ... "CVE": "CVE-2024-12797",
218+ ... "date_published": "2024-02-11",
219+ ... "affected_packages": ["OpenSSL from 1.0.1 to 1.0.1j"],
220+ ... "references": ["https://www.cve.org/CVERecord?id=CVE-2024-12797"],
221+ ... "summary": "Example summary",
222+ ... "severity": "High"
223+ ... }
224+ >>> with patch('dateparser.parse') as mock_dateparser:
225+ ... mock_dateparser.return_value = datetime(2024, 2, 11, tzinfo=timezone.utc)
226+ ... advisory = to_advisory_data(raw_data)
227+ >>> advisory.aliases
228+ ['CVE-2024-12797']
229+ >>> advisory.date_published.isoformat()
230+ '2024-02-11T00:00:00+00:00'
231+ >>> len(advisory.affected_packages)
232+ 1
233+ >>> advisory.references[0].url
234+ 'https://www.cve.org/CVERecord?id=CVE-2024-12797'
235+ """
236+
151237 aliases = [get_item (raw_data , "CVE" )]
152238
153- # published data
154239 date_published = get_item (raw_data , "date_published" )
155240 parsed_date_published = dateparser .parse (date_published , yearfirst = True ).replace (
156241 tzinfo = timezone .utc
157242 )
158243
159- # affected packages
160244 affected_packages = []
161245 affected_package_out = get_item (raw_data , "affected_packages" )
162246 for affected in affected_package_out :
@@ -172,17 +256,14 @@ def to_advisory_data(raw_data) -> AdvisoryData:
172256 )
173257 )
174258
175- # Severity
176259 severity = VulnerabilitySeverity (
177260 system = SCORING_SYSTEMS ["generic_textual" ], value = get_item (raw_data , "severity" )
178261 )
179262
180- # Reference
181263 references = []
182264 for reference in get_item (raw_data , "references" ):
183265 references .append (Reference (severities = [severity ], reference_id = aliases [0 ], url = reference ))
184266
185- # summary
186267 summary = get_item (raw_data , "summary" )
187268
188269 return AdvisoryData (
0 commit comments