Bring the VulnCheck API to your Python applications.
# From PyPi
pip install vulncheck-sdkImportant
Windows users may need to enable Long Path Support
importurllib.requestimportvulncheck_sdkimportos# First let's setup a few variables to help usTOKEN=os.environ["VULNCHECK_API_TOKEN"] # Remember to store your token securely!# Now let's create a configuration objectconfiguration=vulncheck_sdk.Configuration()
configuration.api_key["Bearer"] =TOKEN# Pass that config object to our API client and now...withvulncheck_sdk.ApiClient(configuration) asapi_client:
# We can use two classes to explore the VulnCheck API: EndpointsApi & IndicesApi### EndpointsApi has methods to query every endpoint except `/v3/index`# See the full list of endpoints here: https://docs.vulncheck.com/apiendpoints_client=vulncheck_sdk.EndpointsApi(api_client)
# PURLapi_response=endpoints_client.purl_get("pkg:hex/coherence@0.1.2")
data=V3controllersPurlResponseData=api_response.dataprint(data.cves)
# CPEcpe="cpe:/a:microsoft:internet_explorer:8.0.6001:beta"api_response=endpoints_client.cpe_get(cpe)
forcveinapi_response.data:
print(cve)
# Download a Backupindex="initial-access"api_response=endpoints_client.backup_index_get(index)
file_path=f"{index}.zip"withurllib.request.urlopen(api_response.data[0].url) asresponse:
withopen(file_path, "wb") asfile:
file.write(response.read())
### IndicesApi has methods for each indexindices_client=vulncheck_sdk.IndicesApi(api_client)
# Add query parameters to filter what you needapi_response=indices_client.index_vulncheck_nvd2_get(cve="CVE-2019-19781")
print(api_response.data)Click to View Async Implementation
importasyncioimportosimportaiohttpimportvulncheck_sdk.aioasvcaio# ConfigurationTOKEN=os.environ.get("VULNCHECK_API_TOKEN")
configuration=vcaio.Configuration()
configuration.api_key["Bearer"] =TOKENasyncdefrun_vulnerability_checks():
# Use 'async with' to manage the ApiClient connection poolasyncwithvcaio.ApiClient(configuration) asapi_client:
endpoints_client=vcaio.EndpointsApi(api_client)
indices_client=vcaio.IndicesApi(api_client)
# --- PURL Search ---# 'await' the coroutine to get resultspurl_response=awaitendpoints_client.purl_get("pkg:hex/coherence@0.1.2")
ifpurl_response.data:
print(f"PURL CVEs: {purl_response.data.cves}")
# --- CPE Search ---cpe="cpe:/a:microsoft:internet_explorer:8.0.6001:beta"# 'await' the coroutine to get resultscpe_response=awaitendpoints_client.cpe_get(cpe)
print(f"CPE Results for {cpe}:")
forcveincpe_response.data:
print(f" - {cve}")
# --- Index Query (NVD2) ---# 'await' the coroutine to get resultsnvd_response=awaitindices_client.index_vulncheck_nvd2_get(
cve="CVE-2019-19781"
)
print(f"NVD2 Data: {nvd_response.data}")
# --- Download Backup (Async) ---index_name="initial-access"# 'await' the coroutine to get resultsbackup_response=awaitendpoints_client.backup_index_get(index_name)
ifbackup_response.data:
download_url=backup_response.data[0].urlfile_path=f"{index_name}.zip"print(f"Downloading backup from {download_url}...")
# Use aiohttp (already in your environment) for async downloadasyncwithaiohttp.ClientSession() assession:
asyncwithsession.get(download_url) asresp:
ifresp.status==200:
# 'await' the coroutine to get resultscontent=awaitresp.read()
withopen(file_path, "wb") asf:
f.write(content)
print(f"Saved backup to {file_path}")
if__name__=="__main__":
# Entry point to start the event loopasyncio.run(run_vulnerability_checks())List all advisory feeds and query advisories filtered by feed
importvulncheck_sdkfromvulncheck_sdk.models.search_v4_advisory_return_valueimportSearchV4AdvisoryReturnValuefromvulncheck_sdk.models.search_v4_list_feed_return_valueimportSearchV4ListFeedReturnValueimportosTOKEN=os.environ["VULNCHECK_API_TOKEN"]
configuration=vulncheck_sdk.Configuration()
configuration.api_key["Bearer"] =TOKENwithvulncheck_sdk.ApiClient(configuration) asapi_client:
advisory_client=vulncheck_sdk.AdvisoryApi(api_client)
# List all available advisory feeds (/v4/advisory)feeds: SearchV4ListFeedReturnValue=advisory_client.v4_list_advisory_feeds()
print("Available feeds:")
forfeedinfeeds.data:
print(f"name: {feed.name}")
feed="wolfi"# Query advisories filtered by feed=wolfi (/v4/advisory?feed=wolfi)advisories: SearchV4AdvisoryReturnValue=advisory_client.v4_query_advisories(name=feed)
print(f"{feed.capitalize()} advisories (page 1): {len(advisories.data)} results")
foradvisoryinadvisories.data:
print(f"cve: {advisory.cve_metadata.cve_id}")Click to View Async Implementation
importasyncioimportosimportvulncheck_sdk.aioasvcaiofromvulncheck_sdk.aio.models.search_v4_advisory_return_valueimportSearchV4AdvisoryReturnValuefromvulncheck_sdk.aio.models.search_v4_list_feed_return_valueimportSearchV4ListFeedReturnValueTOKEN=os.environ.get("VULNCHECK_API_TOKEN")
configuration=vcaio.Configuration()
configuration.api_key["Bearer"] =TOKENasyncdefmain():
asyncwithvcaio.ApiClient(configuration) asapi_client:
advisory_client=vcaio.AdvisoryApi(api_client)
# List all available advisory feeds (/v4/advisory)feeds: SearchV4ListFeedReturnValue=awaitadvisory_client.v4_list_advisory_feeds()
print("Available feeds:")
forfeedinfeeds.data:
print(f"name: {feed.name}")
feed="wolfi"# Query advisories filtered by feed=wolfi (/v4/advisory?feed=wolfi)advisories: SearchV4AdvisoryReturnValue=awaitadvisory_client.v4_query_advisories(name=feed)
print(f"{feed.capitalize()} advisories (page 1): {len(advisories.data)} results")
foradvisoryinadvisories.data:
print(f"cve: {advisory.cve_metadata.cve_id}")
if__name__=="__main__":
asyncio.run(main())Download the backup for an index
importurllib.requestimportvulncheck_sdkimportosTOKEN=os.environ["VULNCHECK_API_TOKEN"]
configuration=vulncheck_sdk.Configuration()
configuration.api_key["Bearer"] =TOKENwithvulncheck_sdk.ApiClient(configuration) asapi_client:
endpoints_client=vulncheck_sdk.EndpointsApi(api_client)
index="initial-access"api_response=endpoints_client.backup_index_get(index)
file_path=f"{index}.zip"withurllib.request.urlopen(api_response.data[0].url) asresponse:
withopen(file_path, "wb") asfile:
file.write(response.read())Click to View Async Implementation
importasyncioimportosimporturllib.requestimportvulncheck_sdk.aioasvcaio# ConfigurationTOKEN=os.environ.get("VULNCHECK_API_TOKEN")
configuration=vcaio.Configuration()
configuration.api_key["Bearer"] =TOKENdefdownload_sync(url, file_path):
""" Standard synchronous download using urllib.request. This runs in a separate thread to avoid blocking the event loop. """withurllib.request.urlopen(url) asresponse:
withopen(file_path, "wb") asfile:
file.write(response.read())
asyncdefmain():
# Use 'async with' to manage the connection life-cycleasyncwithvcaio.ApiClient(configuration) asapi_client:
endpoints_client=vcaio.EndpointsApi(api_client)
index="initial-access"# 'await' the coroutine to get the actual response dataapi_response=awaitendpoints_client.backup_index_get(index)
ifnotapi_response.data:
print("No backup URL found.")
returndownload_url=api_response.data[0].urlfile_path=f"{index}.zip"print(f"Downloading {index} via urllib (offloaded to thread)...")
# Use asyncio.to_thread to run the blocking call safely# 'await' the coroutine to get the actual response dataawaitasyncio.to_thread(download_sync, download_url, file_path)
print(f"Successfully saved to {file_path}")
if__name__=="__main__":
asyncio.run(main())List available v4 backups and download a backup by feed name
importosimporttempfileimporturllib.requestfromurllib3.util.retryimportRetryimportvulncheck_sdkfromvulncheck_sdk.models.backup_backup_responseimportBackupBackupResponsefromvulncheck_sdk.models.backup_list_backups_responseimportBackupListBackupsResponseTOKEN=os.environ["VULNCHECK_API_TOKEN"]
configuration=vulncheck_sdk.Configuration()
configuration.api_key["Bearer"] =TOKENconfiguration.retries=Retry(
total=5,
backoff_factor=2,
status_forcelist=[502, 503, 504],
allowed_methods=["GET"],
respect_retry_after_header=True,
)
withvulncheck_sdk.ApiClient(configuration) asapi_client:
backup_client=vulncheck_sdk.BackupApi(api_client)
# List available backups (/v4/backup)available: BackupListBackupsResponse=backup_client.v4_list_backups()
forpotential_backupinavailable.data:
print(f"Found backup: {potential_backup.name}")
# Get backup for the wolfi feed (/v4/backup/wolfi)feed="wolfi"response: BackupBackupResponse=backup_client.v4_get_backup_by_name(feed)
print(response.to_json())
print(f"Downloading {feed} backup")
withtempfile.TemporaryDirectory() astmpdir:
file_path=os.path.join(tmpdir, f"{feed}.zip")
withurllib.request.urlopen(response.url_mrap) asr:
withopen(file_path, "wb") asf:
f.write(r.read())
print(f"Successfully saved to {file_path}")Click to View Async Implementation
importasyncioimportosimporttempfileimporturllib.requestimportvulncheck_sdk.aioasvcaiofromvulncheck_sdk.aio.models.backup_backup_responseimportBackupBackupResponsefromvulncheck_sdk.aio.models.backup_list_backups_responseimport (
BackupListBackupsResponse,
)
TOKEN=os.environ["VULNCHECK_API_TOKEN"]
configuration=vcaio.Configuration()
configuration.api_key["Bearer"] =TOKENconfiguration.retries=5defdownload_sync(url, file_path):
""" Standard synchronous download using urllib.request. This runs in a separate thread to avoid blocking the event loop. """withurllib.request.urlopen(url) asresponse:
withopen(file_path, "wb") asfile:
file.write(response.read())
asyncdefmain():
asyncwithvcaio.ApiClient(configuration) asapi_client:
backup_client=vcaio.BackupApi(api_client)
# List available backups (/v4/backup)available: BackupListBackupsResponse=awaitbackup_client.v4_list_backups()
forpotential_backupinavailable.data:
print(f"Found backup: {potential_backup.name}")
# Get backup for the wolfi feed (/v4/backup/wolfi)feed="wolfi"response: BackupBackupResponse=awaitbackup_client.v4_get_backup_by_name(feed)
print(response.to_json())
print(f"Downloading {feed} backup via urllib (offloaded to thread)...")
withtempfile.TemporaryDirectory() astmpdir:
file_path=os.path.join(tmpdir, f"{feed}.zip")
awaitasyncio.to_thread(download_sync, response.url_mrap, file_path)
print(f"Successfully saved to {file_path}")
if__name__=="__main__":
asyncio.run(main())Get all CPE's related to a CVE
importvulncheck_sdkimportosTOKEN=os.environ["VULNCHECK_API_TOKEN"]
configuration=vulncheck_sdk.Configuration()
configuration.api_key["Bearer"] =TOKENwithvulncheck_sdk.ApiClient(configuration) asapi_client:
endpoints_client=vulncheck_sdk.EndpointsApi(api_client)
cpe="cpe:/a:microsoft:internet_explorer:8.0.6001:beta"api_response=endpoints_client.cpe_get(cpe)
forcveinapi_response.data:
print(cve)Click to View Async Implementation
importasyncioimportosimportvulncheck_sdk.aioasvcaio# ConfigurationTOKEN=os.environ.get("VULNCHECK_API_TOKEN")
configuration=vcaio.Configuration()
configuration.api_key["Bearer"] =TOKENasyncdefget_cpe_vulnerabilities():
# 'async with' to manage the connection life-cycleasyncwithvcaio.ApiClient(configuration) asapi_client:
endpoints_client=vcaio.EndpointsApi(api_client)
cpe="cpe:/a:microsoft:internet_explorer:8.0.6001:beta"# 'await' the coroutine to get the actual response dataapi_response=awaitendpoints_client.cpe_get(cpe)
# Iterate through the resultsifapi_response.data:
forcveinapi_response.data:
print(cve)
else:
print(f"No vulnerabilities found for CPE: {cpe}")
if__name__=="__main__":
# Run the main async entry pointasyncio.run(get_cpe_vulnerabilities())Query VulnCheck-NVD2 for CVE-2019-19781
importvulncheck_sdkimportosTOKEN=os.environ["VULNCHECK_API_TOKEN"]
configuration=vulncheck_sdk.Configuration()
configuration.api_key["Bearer"] =TOKENwithvulncheck_sdk.ApiClient(configuration) asapi_client:
indices_client=vulncheck_sdk.IndicesApi(api_client)
api_response=indices_client.index_vulncheck_nvd2_get(cve="CVE-2019-19781")
print(api_response.data)Click to View Async Implementation
importasyncioimportosimportvulncheck_sdk.aioasvcaio# ConfigurationTOKEN=os.environ.get("VULNCHECK_API_TOKEN")
configuration=vcaio.Configuration()
configuration.api_key["Bearer"] =TOKENasyncdefget_cve_details():
# Use 'async with' for the ApiClientasyncwithvcaio.ApiClient(configuration) asapi_client:
indices_client=vcaio.IndicesApi(api_client)
# 'await' the API callapi_response=awaitindices_client.index_vulncheck_nvd2_get(
cve="CVE-2019-19781"
)
# Access and print the dataifapi_response.data:
print(api_response.data)
else:
print("No data found for the specified CVE.")
if__name__=="__main__":
# Start the async event loopasyncio.run(get_cve_details())Get all available indices
importvulncheck_sdkimportosTOKEN=os.environ["VULNCHECK_API_TOKEN"]
configuration=vulncheck_sdk.Configuration()
configuration.api_key["Bearer"] =TOKENwithvulncheck_sdk.ApiClient(configuration) asapi_client:
endpoints_client=vulncheck_sdk.EndpointsApi(api_client)
api_response=endpoints_client.index_get()
forindexinapi_response.data:
print(index.name)Click to View Async Implementation
importasyncioimportosimportvulncheck_sdk.aioasvcaio# ConfigurationTOKEN=os.environ.get("VULNCHECK_API_TOKEN")
configuration=vcaio.Configuration()
configuration.api_key["Bearer"] =TOKENasyncdeflist_indices():
# Use 'async with' to manage the connection life-cycleasyncwithvcaio.ApiClient(configuration) asapi_client:
endpoints_client=vcaio.EndpointsApi(api_client)
# 'await' the coroutine to get the actual responseapi_response=awaitendpoints_client.index_get()
# Iterate through the resultsifapi_response.data:
print(f"{'Index Name':<30} | {'Description'}")
print("-"*50)
forindexinapi_response.data:
print(f"{index.name:<30}")
else:
print("No indices found.")
if__name__=="__main__":
# 4. Entry point to run the asynchronous event loopasyncio.run(list_indices())Paginate over results for a query to VulnCheck-KEV using cursor
importvulncheck_sdkimportosTOKEN=os.environ["VULNCHECK_API_TOKEN"]
configuration=vulncheck_sdk.Configuration()
configuration.api_key["Bearer"] =TOKENwithvulncheck_sdk.ApiClient(configuration) asapi_client:
indices_client=vulncheck_sdk.IndicesApi(api_client)
api_response=indices_client.index_vulncheck_kev_get(
start_cursor="true",
# `limit` increases the size of each page, making it faster# to download large datasetslimit=300,
)
print(api_response.data)
whileapi_response.meta.next_cursorisnotNone:
api_response=indices_client.index_vulncheck_kev_get(
cursor=api_response.meta.next_cursor
)
print(api_response.data)Click to View Async Implementation
importasyncioimportosimportvulncheck_sdk.aioasvcaio# ConfigurationTOKEN=os.environ.get("VULNCHECK_API_TOKEN")
configuration=vcaio.Configuration()
configuration.api_key["Bearer"] =TOKENasyncdeffetch_kev_data():
# Use 'async with' to properly manage the lifecycle of the async clientasyncwithvcaio.ApiClient(configuration) asapi_client:
indices_client=vcaio.IndicesApi(api_client)
# 'await' the coroutine to get the actual response dataapi_response=awaitindices_client.index_vulncheck_kev_get(
start_cursor="true", limit=300
)
print(f"Fetched {len(api_response.data)} records...")
# Process initial data# (e.g., save to a list or database)# Pagination loopwhileapi_response.metaandapi_response.meta.next_cursor:
print(f"Fetching next page: {api_response.meta.next_cursor}")
# 'await' the coroutine to get the actual response dataapi_response=awaitindices_client.index_vulncheck_kev_get(
cursor=api_response.meta.next_cursor, limit=300
)
ifapi_response.data:
print(f"Fetched {len(api_response.data)} records...")
else:
breakif__name__=="__main__":
# Entry point to run the async event loopasyncio.run(fetch_kev_data())Get the CVE's for a given PURL
importvulncheck_sdkfromvulncheck_sdk.models.v3controllers_purl_response_dataimport (
V3controllersPurlResponseData,
)
importosTOKEN=os.environ["VULNCHECK_API_TOKEN"]
configuration=vulncheck_sdk.Configuration()
configuration.api_key["Bearer"] =TOKENwithvulncheck_sdk.ApiClient(configuration) asapi_client:
endpoints_client=vulncheck_sdk.EndpointsApi(api_client)
purl="pkg:hex/coherence@0.1.2"api_response=endpoints_client.purl_get(purl)
data: V3controllersPurlResponseData=api_response.dataprint(data.cves)Click to View Async Implementation
importasyncioimportosimportvulncheck_sdk.aioasvcaiofromvulncheck_sdk.aio.models.v3controllers_purl_response_dataimport (
V3controllersPurlResponseData,
)
# ConfigurationTOKEN=os.environ.get("VULNCHECK_API_TOKEN")
configuration=vcaio.Configuration()
configuration.api_key["Bearer"] =TOKENasyncdefget_data(client, purl: str):
# Await the client call directlyapi_response=awaitclient.purl_get(purl)
# Access the data attribute from the response objectreturnapi_response.dataasyncdefmain():
asyncwithvcaio.ApiClient(configuration) asapi_client:
endpoints_client=vcaio.EndpointsApi(api_client)
purl="pkg:hex/coherence@0.1.2"# 'await' the async function calldata: V3controllersPurlResponseData=awaitget_data(endpoints_client, purl)
ifdataanddata.cves:
print(f"Found {len(data.cves)} CVEs:")
forcveindata.cves:
print(f"- {cve}")
else:
print("No CVEs found or data is empty.")
if__name__=="__main__":
asyncio.run(main())Please see CONTRIBUTING for details.
If you discover any security related issues, please create an issue.
Development of this project is sponsored by VulnCheck learn more about us!
Apache License 2.0. Please see License File for more information.
