Merge branch 'main' into api-changes

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pyvulnerabilitylookup"
-version = "2.1.0"
+version = "2.1.1"
 description = "Python CLI and module for Vulnerability Lookup"
 authors = ["Raphaël Vinot <raphael.vinot@circl.lu>"]
 license = "BSD-3-Clause"

pyvulnerabilitylookup/api.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import functools
 
 from datetime import date, datetime
 from importlib.metadata import version
@@ -12,6 +13,9 @@ from urllib.parse import urljoin, urlparse
 
 import requests
 
+from urllib3.util import Retry
+from requests.adapters import HTTPAdapter
+
 
 def enable_full_debug() -> None:
     import http.client as http_client
@@ -46,6 +50,8 @@ class PyVulnerabilityLookup():
         self.session.headers['Content-Type'] = 'application/json'
         if proxies:
             self.session.proxies.update(proxies)
+        retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
+        self.session.mount('https://', HTTPAdapter(max_retries=retries))
 
     def set_apikey(self, apikey: str) -> None:
         '''Set the API key to use for the requests'''
@@ -186,8 +192,15 @@ class PyVulnerabilityLookup():
         :param vuln_id: The vulnerability ID to get comments of
         :param author: The author of the comment(s)
         '''
+        params = {}
+        if uuid:
+            params['uuid'] = uuid
+        if vuln_id:
+            params['vuln_id'] = vuln_id
+        if author:
+            params['author'] = author
         r = self.session.get(urljoin(self.root_url, str(PurePosixPath('api', 'comment'))),
-                             params={'uuid': uuid, 'vuln_id': vuln_id, 'author': author})
+                             params=params)
         return r.json()
 
     def get_comment(self, comment_uuid: str) -> dict[str, Any]:
@@ -327,8 +340,7 @@ class PyVulnerabilityLookup():
 
         :param user_id: The user ID
         '''
-        r = self.session.delete(urljoin(self.root_url, str(PurePosixPath('api', 'user', user_id)))
-                                )
+        r = self.session.delete(urljoin(self.root_url, str(PurePosixPath('api', 'user', user_id))))
         return r.status_code
 
     # #### Sightings ####

tests/test_web.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 
 import unittest
+import uuid
 import time
 import os
 
@@ -133,6 +134,10 @@ class TestPublic(unittest.TestCase):
             comments["data"][0]["uuid"], "a309d024-2714-4a81-a425-60f83f6d5740"
         )
 
+        comment = self.client.get_comment("a309d024-2714-4a81-a425-60f83f6d5740")
+        self.assertEqual(
+            comment["uuid"], "a309d024-2714-4a81-a425-60f83f6d5740"
+        )
         comments = self.client.get_comments(uuid="a309d024-2714-4a81-a425-60f83f6d5740")
         self.assertTrue(len(comments["data"]) == 1)
         self.assertEqual(
@@ -296,6 +301,8 @@ class TestPublic(unittest.TestCase):
         user = self.client.create_user(name='test Name', login='alan',
                                        organisation='test Organization', email='test@testorg.lu')
         self.assertTrue(user)
+        self.assertTrue('id' in user, user)
+        uid = user['id']
         self.assertTrue('login' in user, user)
         self.assertTrue('apikey' in user, user)
         self.assertTrue('is_commenter' in user, user)
@@ -311,8 +318,14 @@ class TestPublic(unittest.TestCase):
         deleted_comment = self.client.delete_comment(new_comment_uuid)
         self.assertTrue(deleted_comment < 300)
 
+        self.client.set_apikey(self.admin_token)
+        deleted_user = self.client.delete_user(str(uid))
+        self.assertTrue(deleted_user < 300)
+
     # test Sightings
-    def test_sightings(self) -> None:
+    def test_sightings_public(self) -> None:
+        if not self.public_test:
+            return None
         sighting_cve = self.client.get_sighting('6febe45d-d8de-4df7-b3ba-6cf7acd2e2b5')
         self.assertTrue(sighting_cve)
         self.assertTrue('uuid' in sighting_cve)
@@ -343,3 +356,30 @@ class TestPublic(unittest.TestCase):
         self.assertTrue(sighting_cve_list)
         self.assertTrue('data' in sighting_cve_list)
         self.assertTrue(len(sighting_cve_list['data']) > 0)
+
+    def test_sightings_local(self) -> None:
+        if not self.admin_token:
+            # this test is only working if the admin token is set
+            return None
+
+        u1 = str(uuid.uuid4())
+        sighting = self.client.create_sighting(
+            sighting={
+                "vulnerability": "CVE-2024-20401",
+                "source": u1,
+                "type": "seen"
+            }
+        )
+        self.assertTrue(sighting)
+        print(sighting)
+        s = self.client.get_sighting(sighting['data'][0]['uuid'])
+        self.assertTrue('uuid' in s)
+        self.assertTrue('vulnerability' in s)
+        self.assertTrue('source' in s)
+        self.assertTrue('type' in s)
+        self.assertEqual(s['source'], u1)
+
+        u2 = str(uuid.uuid4())
+        sighting = self.client.create_sighting(source=u2, sighting_type='seen', vulnerability='CVE-2024-20401')
+        s = self.client.get_sighting(sighting['data'][0]['uuid'])
+        self.assertEqual(s['source'], u2)