Resources

Each namespace on the client (v.search, v.audit, …) is a resource. The classes below are the synchronous resources; the async client exposes an Async* mirror of each, with the same non-iterator methods as awaitable coroutines. The auto-paginating streaming iterators are renamed with an a prefix (aiter_query, aiter_collection) and are async generators consumed with async for, not coroutines.

Search

vulners._resources._sync.search.Search

Search(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

Search the Vulners database.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

query

query(query: str, *, limit: int = 20, offset: int = 0, fields: Sequence[str] | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> SearchPage[Bulletin]

Search using Lucene query syntax.

Parameters:

Name	Type	Description	Default
query	str	A Vulners Lucene query (see https://vulners.com/help).	required
limit	int	Maximum number of documents to return in this page.	20
offset	int	Number of documents to skip.	0
fields	Sequence[str] | NotGiven	Restrict the returned fields. When omitted, the compact :data:DEFAULT_SEARCH_FIELDS projection is requested; the heavy document fields (sourceData, description) are opt-in — pass fields=["*"] for whole documents or list fields explicitly.	not_given

Returns:

Type	Description
SearchPage[Bulletin]	A cursor-aware :class:vulners._pagination.SearchPage of :class:Bulletin;
SearchPage[Bulletin]	.total is the total match count and iterating the page walks
SearchPage[Bulletin]	further pages up to the 10000-document window.

Raises:

Type	Description
SearchWindowExceeded	offset is at or beyond the 10000-document window; use the archive API to page further.

Source code in src/vulners/_resources/_sync/search.py

def query(
    self,
    query: str,
    *,
    limit: int = 20,
    offset: int = 0,
    fields: Sequence[str] | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> SearchPage[Bulletin]:
    """Search using Lucene query syntax.

    Args:
        query: A Vulners Lucene query (see https://vulners.com/help).
        limit: Maximum number of documents to return in this page.
        offset: Number of documents to skip.
        fields: Restrict the returned fields. When omitted, the compact
            :data:`DEFAULT_SEARCH_FIELDS` projection is requested; the heavy
            document fields (``sourceData``, ``description``) are opt-in —
            pass ``fields=["*"]`` for whole documents or list fields
            explicitly.

    Returns:
        A cursor-aware :class:`vulners._pagination.SearchPage` of :class:`Bulletin`;
        ``.total`` is the total match count and iterating the page walks
        further pages up to the 10000-document window.

    Raises:
        SearchWindowExceeded: ``offset`` is at or beyond the 10000-document
            window; use the archive API to page further.
    """
    if not query or not query.strip():
        raise ValueError("query must not be empty")
    if limit < 1:
        raise ValueError(f"limit must be >= 1, got {limit}")
    if offset < 0:
        raise ValueError(f"offset must be >= 0, got {offset}")
    if offset >= SEARCH_WINDOW:
        raise SearchWindowExceeded(
            f"offset must be less than {SEARCH_WINDOW} (got {offset}); the search "
            "window is capped at 10000 documents. Use the archive API to retrieve more."
        )
    size = min(limit, SEARCH_WINDOW - offset)
    body: dict[str, Any] = {"query": query, "size": size, "skip": offset}
    body["fields"] = list(DEFAULT_SEARCH_FIELDS if isinstance(fields, NotGiven) else fields)

    def _fetch(next_offset: int, next_size: int) -> SearchPage[Bulletin]:
        return self.query(
            query, limit=next_size, offset=next_offset, fields=fields, timeout=timeout
        )

    def _build(data: Any) -> SearchPage[Bulletin]:
        hits = data.get("search", []) if isinstance(data, dict) else []
        rows = [
            construct_bulletin(hit.get("_source", {}))
            for hit in hits
            if isinstance(hit, dict)
        ]
        total = data.get("total") if isinstance(data, dict) else None
        return SearchPage(data=rows, total=total, offset=offset, limit=size, fetch=_fetch)

    return self._request(_SEARCH_SPEC, cast=_build, body=body, timeout=timeout)

iter_query

iter_query(query: str, *, page_size: int = 100, fields: Sequence[str] | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> Iterator[Bulletin]

Iterate every matching :class:Bulletin, auto-paginating.

Fetches pages of page_size documents and yields their rows lazily, stopping at the 10000-document search window.

Parameters:

Name	Type	Description	Default
query	str	A Vulners Lucene query (see https://vulners.com/help).	required
page_size	int	Number of documents to fetch per underlying page request.	100
fields	Sequence[str] | NotGiven	Restrict the returned fields, as in :meth:query. When omitted the compact :data:DEFAULT_SEARCH_FIELDS projection is used; the heavy document fields (sourceData, description) are opt-in via fields=["*"] or an explicit list.	not_given

Yields:

Type	Description
Bulletin	Each matching :class:Bulletin, in result order, across every page
Bulletin	up to the 10000-document search window.

Source code in src/vulners/_resources/_sync/search.py

def iter_query(
    self,
    query: str,
    *,
    page_size: int = 100,
    fields: Sequence[str] | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Iterator[Bulletin]:
    """Iterate every matching :class:`Bulletin`, auto-paginating.

    Fetches pages of ``page_size`` documents and yields their rows lazily,
    stopping at the 10000-document search window.

    Args:
        query: A Vulners Lucene query (see https://vulners.com/help).
        page_size: Number of documents to fetch per underlying page request.
        fields: Restrict the returned fields, as in :meth:`query`. When
            omitted the compact :data:`DEFAULT_SEARCH_FIELDS` projection is
            used; the heavy document fields (``sourceData``, ``description``)
            are opt-in via ``fields=["*"]`` or an explicit list.

    Yields:
        Each matching :class:`Bulletin`, in result order, across every page
        up to the 10000-document search window.
    """
    page = self.query(query, limit=page_size, offset=0, fields=fields, timeout=timeout)
    while True:
        for row in page.data:
            yield row
        if not page.has_next_page():
            return
        page = page.next_page()

get_multiple_bulletins

get_multiple_bulletins(ids: Sequence[str], *, fields: Sequence[str] | NotGiven = not_given, references: bool = False) -> dict[str, Bulletin]

Fetch several documents by id, keyed by id.

Parameters:

Name	Type	Description	Default
ids	Sequence[str]	Document ids to fetch (e.g. CVE-2021-44228). The returned mapping is keyed by these ids; ids with no matching document are absent from the result.	required
fields	Sequence[str] | NotGiven	Restrict the returned fields on each document; the server default projection is used when omitted.	not_given
references	bool	When True, also resolve and include documents the requested bulletins reference/cross-link (e.g. the exploits and advisories tied to a CVE); defaults to False.	False

Returns:

Type	Description
dict[str, Bulletin]	A mapping of id to the family-specific :class:Bulletin.

Source code in src/vulners/_resources/_sync/search.py

def get_multiple_bulletins(
    self,
    ids: Sequence[str],
    *,
    fields: Sequence[str] | NotGiven = not_given,
    references: bool = False,
) -> dict[str, Bulletin]:
    """Fetch several documents by id, keyed by id.

    Args:
        ids: Document ids to fetch (e.g. ``CVE-2021-44228``). The returned
            mapping is keyed by these ids; ids with no matching document are
            absent from the result.
        fields: Restrict the returned fields on each document; the server
            default projection is used when omitted.
        references: When ``True``, also resolve and include documents the
            requested bulletins reference/cross-link (e.g. the exploits and
            advisories tied to a CVE); defaults to ``False``.

    Returns:
        A mapping of id to the family-specific :class:`Bulletin`.
    """
    body: dict[str, Any] = {"id": list(ids), "references": references}
    self._set(body, "fields", fields, list)

    def _cast(data: Any) -> dict[str, Bulletin]:
        docs = data.get("documents", {}) if isinstance(data, dict) else {}
        return {
            key: construct_bulletin(value)
            for key, value in docs.items()
            if isinstance(value, dict)
        }

    return self._request(_LOOKUP_SPEC, cast=_cast, body=body)

get_bulletin

get_bulletin(id: str, *, fields: Sequence[str] | NotGiven = not_given) -> Bulletin | None

Fetch a single document by id, or None if it does not exist.

Parameters:

Name	Type	Description	Default
id	str	The document id to fetch (e.g. CVE-2021-44228).	required
fields	Sequence[str] | NotGiven	Restrict the returned fields; the server default projection is used when omitted.	not_given

Returns:

Type	Description
Bulletin | None	The family-specific :class:Bulletin, or None if no document
Bulletin | None	matches id.

Source code in src/vulners/_resources/_sync/search.py

def get_bulletin(
    self,
    id: str,
    *,
    fields: Sequence[str] | NotGiven = not_given,
) -> Bulletin | None:
    """Fetch a single document by id, or ``None`` if it does not exist.

    Args:
        id: The document id to fetch (e.g. ``CVE-2021-44228``).
        fields: Restrict the returned fields; the server default projection
            is used when omitted.

    Returns:
        The family-specific :class:`Bulletin`, or ``None`` if no document
        matches ``id``.
    """
    body: dict[str, Any] = {"id": [id], "references": False}
    self._set(body, "fields", fields, list)

    def _cast(data: Any) -> Bulletin | None:
        docs = data.get("documents", {}) if isinstance(data, dict) else {}
        raw = docs.get(id)
        return construct_bulletin(raw) if isinstance(raw, dict) else None

    return self._request(_LOOKUP_SPEC, cast=_cast, body=body)

exploits

exploits(query: str, *, lucene: bool = False, limit: int = 20, offset: int = 0, fields: Sequence[str] | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> SearchPage[Bulletin]

Search for public exploits matching a query.

Restricts a Lucene search to the exploit bulletin family. A bare CVE id (e.g. CVE-2021-44228) is phrase-quoted automatically so it matches as a single token.

Parameters:

Name	Type	Description	Default
query	str	A CVE id, product name, or Lucene fragment.	required
lucene	bool	Treat query as raw Lucene: no CVE auto-quoting, only the exploit-family filter is applied around it.	False
limit	int	Maximum number of documents to return in this page.	20
offset	int	Number of documents to skip.	0
fields	Sequence[str] | NotGiven	Restrict the returned fields, as in :meth:query.	not_given

Returns:

Type	Description
SearchPage[Bulletin]	A cursor-aware :class:vulners._pagination.SearchPage of :class:Bulletin.

Raises:

Type	Description
SearchWindowExceeded	offset is at or beyond the 10000-document window.

Source code in src/vulners/_resources/_sync/search.py

def exploits(
    self,
    query: str,
    *,
    lucene: bool = False,
    limit: int = 20,
    offset: int = 0,
    fields: Sequence[str] | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> SearchPage[Bulletin]:
    """Search for public exploits matching a query.

    Restricts a Lucene search to the ``exploit`` bulletin family. A bare
    CVE id (e.g. ``CVE-2021-44228``) is phrase-quoted automatically so it
    matches as a single token.

    Args:
        query: A CVE id, product name, or Lucene fragment.
        lucene: Treat ``query`` as raw Lucene: no CVE auto-quoting, only
            the exploit-family filter is applied around it.
        limit: Maximum number of documents to return in this page.
        offset: Number of documents to skip.
        fields: Restrict the returned fields, as in :meth:`query`.

    Returns:
        A cursor-aware :class:`vulners._pagination.SearchPage` of :class:`Bulletin`.

    Raises:
        SearchWindowExceeded: ``offset`` is at or beyond the 10000-document
            window.
    """
    return self.query(
        exploit_search_query(query, lucene=lucene),
        limit=limit,
        offset=offset,
        fields=fields,
        timeout=timeout,
    )

collections

collections(*, timeout: float | Timeout | NotGiven = not_given) -> Any

List every collection available in Vulners.

Returns:

Type	Description
Any	A list of collection descriptors, each with type, count,
Any	description and last_updated, sorted by count descending.

Source code in src/vulners/_resources/_sync/search.py

def collections(
    self,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """List every collection available in Vulners.

    Returns:
        A list of collection descriptors, each with ``type``, ``count``,
        ``description`` and ``last_updated``, sorted by count descending.
    """
    return self._request(_COLLECTIONS_SPEC, timeout=timeout)

autocomplete

autocomplete(query: str, *, timeout: float | Timeout | NotGiven = not_given) -> list[str | list[str]]

Return possible completions for a partial Lucene query.

Same endpoint as :meth:Misc.query_autocomplete.

Parameters:

Name	Type	Description	Default
query	str	The partial Lucene query to complete.	required

Returns:

Type	Description
list[str | list[str]]	Ordered completions: each element is a completion string, or a
list[str | list[str]]	list[str] when the server groups several related completions.

Source code in src/vulners/_resources/_sync/search.py

def autocomplete(
    self,
    query: str,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> list[str | list[str]]:
    """Return possible completions for a partial Lucene query.

    Same endpoint as :meth:`Misc.query_autocomplete`.

    Args:
        query: The partial Lucene query to complete.

    Returns:
        Ordered completions: each element is a completion string, or a
        ``list[str]`` when the server groups several related completions.
    """
    return self._request(
        _AUTOCOMPLETE, cast=_autocomplete, body={"query": query}, timeout=timeout
    )

suggest

suggest(field_name: str, *, type: Literal['distinct'] = 'distinct', timeout: float | Timeout | NotGiven = not_given) -> Any

Return distinct value suggestions for a document field.

Same endpoint as :meth:Misc.get_suggestion.

Parameters:

Name	Type	Description	Default
field_name	str	The document field to suggest values for.	required
type	Literal['distinct']	Suggestion type; only "distinct" is supported.	'distinct'

Source code in src/vulners/_resources/_sync/search.py

def suggest(
    self,
    field_name: str,
    *,
    type: Literal["distinct"] = "distinct",
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Return distinct value suggestions for a document field.

    Same endpoint as :meth:`Misc.get_suggestion`.

    Args:
        field_name: The document field to suggest values for.
        type: Suggestion type; only ``"distinct"`` is supported.
    """
    body = {"fieldName": field_name, "type": type}
    return self._request(_SUGGEST, body=body, timeout=timeout)

cpe

cpe(product: str, *, vendor: str | NotGiven = not_given, size: int | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> Any

Search for CPE strings matching a product (and optional vendor).

Same endpoint as :meth:Misc.search_cpe.

Parameters:

Name	Type	Description	Default
product	str	Product string to search a CPE for.	required
vendor	str | NotGiven	Optional vendor to narrow the match.	not_given
size	int | NotGiven	Maximum number of results (0..10000, 0 means all).	not_given

Source code in src/vulners/_resources/_sync/search.py

def cpe(
    self,
    product: str,
    *,
    vendor: str | NotGiven = not_given,
    size: int | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Search for CPE strings matching a product (and optional vendor).

    Same endpoint as :meth:`Misc.search_cpe`.

    Args:
        product: Product string to search a CPE for.
        vendor: Optional vendor to narrow the match.
        size: Maximum number of results (0..10000, ``0`` means all).
    """
    body: dict[str, Any] = {"product": product}
    self._set(body, "vendor", vendor)
    self._set(body, "size", size)
    return self._request(_SEARCH_CPE, body=body, timeout=timeout)

web_vulns

web_vulns(*, timeout: float | Timeout | NotGiven = not_given) -> Any

Return the Vulners web-application (burp) detection rule set.

Same endpoint as :meth:Misc.get_web_application_rules.

Returns:

Type	Description
Any	The web-application (burp) detection rule set: the software
Any	signatures used to detect web applications and their known
Any	vulnerabilities.

Source code in src/vulners/_resources/_sync/search.py

def web_vulns(
    self,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Return the Vulners web-application (burp) detection rule set.

    Same endpoint as :meth:`Misc.get_web_application_rules`.

    Returns:
        The web-application (burp) detection rule set: the software
        signatures used to detect web applications and their known
        vulnerabilities.
    """
    return self._request(_BURP_RULES, timeout=timeout)

Documents

vulners._resources._sync.documents.Documents

Documents(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

Fetch Vulners documents (bulletins) by id.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

get

get(id: str, *, references: bool = False, fields: Sequence[str] | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> Bulletin | None

Fetch a single document by id, or None if it does not exist.

Parameters:

Name	Type	Description	Default
id	str	The document id to fetch (e.g. CVE-2021-44228).	required
references	bool	Also resolve the documents this bulletin references server-side; fetch them with :meth:references.	False
fields	Sequence[str] | NotGiven	Restrict the returned fields; the server default projection is used when omitted.	not_given

Returns:

Type	Description
Bulletin | None	The family-specific :class:Bulletin, or None if no document
Bulletin | None	matches id.

Source code in src/vulners/_resources/_sync/documents.py

def get(
    self,
    id: str,
    *,
    references: bool = False,
    fields: Sequence[str] | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Bulletin | None:
    """Fetch a single document by id, or ``None`` if it does not exist.

    Args:
        id: The document id to fetch (e.g. ``CVE-2021-44228``).
        references: Also resolve the documents this bulletin references
            server-side; fetch them with :meth:`references`.
        fields: Restrict the returned fields; the server default projection
            is used when omitted.

    Returns:
        The family-specific :class:`Bulletin`, or ``None`` if no document
        matches ``id``.
    """
    body: dict[str, Any] = {"id": [id], "references": references}
    self._set(body, "fields", fields, list)

    def _cast(data: Any) -> Bulletin | None:
        docs = data.get("documents", {}) if isinstance(data, dict) else {}
        raw = docs.get(id)
        return construct_bulletin(raw) if isinstance(raw, dict) else None

    return self._request(_LOOKUP_SPEC, cast=_cast, body=body, timeout=timeout)

get_many

get_many(ids: Sequence[str], *, references: bool = False, fields: Sequence[str] | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> dict[str, Bulletin]

Fetch several documents by id, keyed by id.

Parameters:

Name	Type	Description	Default
ids	Sequence[str]	Document ids to fetch. Ids with no matching document are absent from the result.	required
references	bool	Also resolve the documents these bulletins reference server-side; fetch them with :meth:references.	False
fields	Sequence[str] | NotGiven	Restrict the returned fields on each document; the server default projection is used when omitted.	not_given

Returns:

Type	Description
dict[str, Bulletin]	A mapping of id to the family-specific :class:Bulletin.

Source code in src/vulners/_resources/_sync/documents.py

def get_many(
    self,
    ids: Sequence[str],
    *,
    references: bool = False,
    fields: Sequence[str] | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, Bulletin]:
    """Fetch several documents by id, keyed by id.

    Args:
        ids: Document ids to fetch. Ids with no matching document are
            absent from the result.
        references: Also resolve the documents these bulletins reference
            server-side; fetch them with :meth:`references`.
        fields: Restrict the returned fields on each document; the server
            default projection is used when omitted.

    Returns:
        A mapping of id to the family-specific :class:`Bulletin`.
    """
    body: dict[str, Any] = {"id": list(ids), "references": references}
    self._set(body, "fields", fields, list)

    def _cast(data: Any) -> dict[str, Bulletin]:
        docs = data.get("documents", {}) if isinstance(data, dict) else {}
        return {
            key: construct_bulletin(value)
            for key, value in docs.items()
            if isinstance(value, dict)
        }

    return self._request(_LOOKUP_SPEC, cast=_cast, body=body, timeout=timeout)

references

references(id: str, *, fields: Sequence[str] | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> dict[str, list[Bulletin]]

Fetch the documents a bulletin references, grouped by source type.

Parameters:

Name	Type	Description	Default
id	str	The document whose references to resolve.	required
fields	Sequence[str] | NotGiven	Restrict the returned fields on each referenced document; the server default projection is used when omitted.	not_given

Returns:

Type	Description
dict[str, list[Bulletin]]	A mapping of source type (e.g. "nessus", "zdt") to the
dict[str, list[Bulletin]]	referencing/referenced :class:Bulletin documents; empty when the
dict[str, list[Bulletin]]	document does not exist or has no references.

Source code in src/vulners/_resources/_sync/documents.py

def references(
    self,
    id: str,
    *,
    fields: Sequence[str] | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, list[Bulletin]]:
    """Fetch the documents a bulletin references, grouped by source type.

    Args:
        id: The document whose references to resolve.
        fields: Restrict the returned fields on each referenced document;
            the server default projection is used when omitted.

    Returns:
        A mapping of source type (e.g. ``"nessus"``, ``"zdt"``) to the
        referencing/referenced :class:`Bulletin` documents; empty when the
        document does not exist or has no references.
    """
    body: dict[str, Any] = {"id": [id], "references": True}
    self._set(body, "fields", fields, list)

    def _cast(data: Any) -> dict[str, list[Bulletin]]:
        refs = data.get("references", {}) if isinstance(data, dict) else {}
        groups = refs.get(id, {}) if isinstance(refs, dict) else {}
        if not isinstance(groups, dict):
            return {}
        return {
            source: [construct_bulletin(doc) for doc in docs if isinstance(doc, dict)]
            for source, docs in groups.items()
            if isinstance(docs, list)
        }

    return self._request(_LOOKUP_SPEC, cast=_cast, body=body, timeout=timeout)

history

history(id: str, *, timeout: float | Timeout | NotGiven = not_given) -> list[dict[str, Any]]

Read the per-field edition history of a bulletin.

Parameters:

Name	Type	Description	Default
id	str	The bulletin id (e.g. CVE-2024-23622).	required

Returns:

Type	Description
list[dict[str, Any]]	A list of {"field", "value", "published"} entries — one per
list[dict[str, Any]]	recorded field edition, newest first.

Source code in src/vulners/_resources/_sync/documents.py

def history(
    self,
    id: str,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> list[dict[str, Any]]:
    """Read the per-field edition history of a bulletin.

    Args:
        id: The bulletin id (e.g. ``CVE-2024-23622``).

    Returns:
        A list of ``{"field", "value", "published"}`` entries — one per
        recorded field edition, newest first.
    """
    return self._request(_HISTORY_SPEC, body={"id": id}, timeout=timeout)

Audit

vulners._resources._sync.audit.Audit

Audit(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

Audit software inventories and identifiers against Vulners intelligence.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

packages cached property

packages: AuditPackages

Package-manager manifest audits (pip/npm/maven/...).

supported_os

supported_os(*, timeout: float | Timeout | NotGiven = not_given) -> dict[str, str]

List the operating systems accepted by the Linux-package audits.

Returns:

Type	Description
dict[str, str]	A mapping of OS short name (e.g. "ubuntu", "rhel") to the
dict[str, str]	shell command Vulners recommends for enumerating that OS's
dict[str, str]	installed packages.

Source code in src/vulners/_resources/_sync/audit.py

def supported_os(
    self,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, str]:
    """List the operating systems accepted by the Linux-package audits.

    Returns:
        A mapping of OS short name (e.g. ``"ubuntu"``, ``"rhel"``) to the
        shell command Vulners recommends for enumerating that OS's
        installed packages.
    """
    return self._request(_SUPPORTED_OS, timeout=timeout)

software

software(software: Sequence[AuditItem | str], *, match: Literal['partial', 'full'] = 'partial', fields: Sequence[str] | NotGiven = not_given, config: Sequence[str] | NotGiven = not_given, catalog: Literal['official', 'extended'] = 'official', cvelist_metrics: bool | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> list[dict[str, Any]]

Audit a list of software (CPE dicts or strings) for vulnerabilities.

Each result carries the input, the matched CPE criteria, the fixed_version that resolves every finding, and the vulnerabilities. The applied field projection is echoed in the X-Vulners-Applied-Options response header (read it via :attr:with_raw_response).

Parameters:

Name	Type	Description	Default
software	Sequence[AuditItem | str]	Entries as :class:AuditItem dicts ({"product": ...}) or plain CPE-like strings.	required
match	Literal['partial', 'full']	"partial" or "full" CPE matching.	'partial'
fields	Sequence[str] | NotGiven	Vulnerability fields to include; server default when omitted.	not_given
config	Sequence[str] | NotGiven	Optional configuration entries.	not_given
catalog	Literal['official', 'extended']	"official" or "extended" product catalog.	'official'
cvelist_metrics	bool | NotGiven	When True, enrich each finding with per-CVE cvelist and cvelistMetrics (CVSS/EPSS per CVE).	not_given

Source code in src/vulners/_resources/_sync/audit.py

def software(
    self,
    software: Sequence[AuditItem | str],
    *,
    match: Literal["partial", "full"] = "partial",
    fields: Sequence[str] | NotGiven = not_given,
    config: Sequence[str] | NotGiven = not_given,
    catalog: Literal["official", "extended"] = "official",
    cvelist_metrics: bool | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> list[dict[str, Any]]:
    """Audit a list of software (CPE dicts or strings) for vulnerabilities.

    Each result carries the input, the matched CPE criteria, the
    ``fixed_version`` that resolves every finding, and the vulnerabilities.
    The applied field projection is echoed in the ``X-Vulners-Applied-Options``
    response header (read it via :attr:`with_raw_response`).

    Args:
        software: Entries as :class:`AuditItem` dicts (``{"product": ...}``)
            or plain CPE-like strings.
        match: ``"partial"`` or ``"full"`` CPE matching.
        fields: Vulnerability fields to include; server default when omitted.
        config: Optional configuration entries.
        catalog: ``"official"`` or ``"extended"`` product catalog.
        cvelist_metrics: When ``True``, enrich each finding with per-CVE
            ``cvelist`` and ``cvelistMetrics`` (CVSS/EPSS per CVE).
    """
    body: dict[str, Any] = {
        "software": _require_count(software, "software"),
        "match": match,
        "catalog": catalog,
    }
    self._set(body, "fields", fields, list)
    self._set(body, "config", config, list)
    self._set(body, "cvelistMetrics", cvelist_metrics)
    return self._request(_SOFTWARE, body=body, timeout=timeout)

host

host(software: Sequence[AuditItem | str], *, application: AuditItem | str | NotGiven = not_given, operating_system: AuditItem | str | NotGiven = not_given, hardware: AuditItem | str | NotGiven = not_given, match: Literal['partial', 'full'] = 'partial', fields: Sequence[str] | NotGiven = not_given, config: Sequence[str] | NotGiven = not_given, catalog: Literal['official', 'extended'] = 'official', cvelist_metrics: bool | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> list[dict[str, Any]]

Audit a whole host: its software plus optional application, OS and hardware CPEs.

Like :meth:software, but also accepts CPEs describing the host's application, operating system and hardware; at least one of application/operating_system/hardware must be set.

Parameters:

Name	Type	Description	Default
software	Sequence[AuditItem | str]	Installed software as :class:AuditItem dicts ({"product": ...}) or plain CPE-like strings.	required
application	AuditItem | str | NotGiven	Optional application CPE narrowing the audit.	not_given
operating_system	AuditItem | str | NotGiven	Optional operating-system CPE.	not_given
hardware	AuditItem | str | NotGiven	Optional hardware CPE.	not_given
match	Literal['partial', 'full']	"partial" or "full" CPE matching.	'partial'
fields	Sequence[str] | NotGiven	Vulnerability fields to include; server default when omitted.	not_given
config	Sequence[str] | NotGiven	Optional configuration entries.	not_given
catalog	Literal['official', 'extended']	"official" or "extended" product catalog.	'official'
cvelist_metrics	bool | NotGiven	When True, enrich each finding with per-CVE cvelist and cvelistMetrics (CVSS/EPSS per CVE).	not_given

Source code in src/vulners/_resources/_sync/audit.py

def host(
    self,
    software: Sequence[AuditItem | str],
    *,
    application: AuditItem | str | NotGiven = not_given,
    operating_system: AuditItem | str | NotGiven = not_given,
    hardware: AuditItem | str | NotGiven = not_given,
    match: Literal["partial", "full"] = "partial",
    fields: Sequence[str] | NotGiven = not_given,
    config: Sequence[str] | NotGiven = not_given,
    catalog: Literal["official", "extended"] = "official",
    cvelist_metrics: bool | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> list[dict[str, Any]]:
    """Audit a whole host: its software plus optional application, OS and hardware CPEs.

    Like :meth:`software`, but also accepts CPEs describing the host's
    application, operating system and hardware; at least one of
    ``application``/``operating_system``/``hardware`` must be set.

    Args:
        software: Installed software as :class:`AuditItem` dicts
            (``{"product": ...}``) or plain CPE-like strings.
        application: Optional application CPE narrowing the audit.
        operating_system: Optional operating-system CPE.
        hardware: Optional hardware CPE.
        match: ``"partial"`` or ``"full"`` CPE matching.
        fields: Vulnerability fields to include; server default when omitted.
        config: Optional configuration entries.
        catalog: ``"official"`` or ``"extended"`` product catalog.
        cvelist_metrics: When ``True``, enrich each finding with per-CVE
            ``cvelist`` and ``cvelistMetrics`` (CVSS/EPSS per CVE).
    """
    body: dict[str, Any] = {
        "software": _require_count(software, "software"),
        "match": match,
        "catalog": catalog,
    }
    self._set(body, "application", application)
    self._set(body, "operating_system", operating_system)
    self._set(body, "hardware", hardware)
    self._set(body, "fields", fields, list)
    self._set(body, "config", config, list)
    self._set(body, "cvelistMetrics", cvelist_metrics)
    return self._request(_HOST, body=body, timeout=timeout)

os_audit

os_audit(os: str, version: str, packages: Sequence[str], *, timeout: float | Timeout | NotGiven = not_given) -> dict[str, Any]

Audit an OS package list (legacy v3 endpoint; prefer :meth:linux_audit).

Parameters:

Name	Type	Description	Default
os	str	OS name, e.g. "ubuntu", "debian", "rhel".	required
version	str	OS version.	required
packages	Sequence[str]	Installed packages, one per entry.	required

Source code in src/vulners/_resources/_sync/audit.py

def os_audit(
    self,
    os: str,
    version: str,
    packages: Sequence[str],
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, Any]:
    """Audit an OS package list (legacy v3 endpoint; prefer :meth:`linux_audit`).

    Args:
        os: OS name, e.g. ``"ubuntu"``, ``"debian"``, ``"rhel"``.
        version: OS version.
        packages: Installed packages, one per entry.
    """
    body = {"os": os, "version": version, "package": list(packages)}
    return self._request(_OS_AUDIT, body=body, timeout=timeout)

linux_audit

linux_audit(os_name: str, os_version: str, packages: Sequence[str], *, os_arch: str | None = None, include_unofficial: bool = False, include_candidates: bool = False, include_any_version: bool = False, cvelist_metrics: bool = False, fields: Sequence[str] | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> dict[str, Any]

Audit RPM/DEB/APK package lists for a Linux host.

Each issue carries the package, its fixedVersion and fixedPackage, and the applicableAdvisories that match it. The result also reports the appliedOptions that took effect and any warnings (for example an unsupported fields option).

Parameters:

Name	Type	Description	Default
os_name	str	OS name or id (ubuntu, debian, rhel, alpine...).	required
os_version	str	OS version.	required
packages	Sequence[str]	Installed packages (1..2500 entries).	required
os_arch	str | None	Default architecture for packages.	None
include_unofficial	bool	Include unofficial packages.	False
include_candidates	bool	Include candidate vulnerabilities.	False
include_any_version	bool	Include any version vulnerabilities.	False
cvelist_metrics	bool	Add cvelist metrics.	False
fields	Sequence[str] | NotGiven	Advisory enrichment options to apply; this endpoint supports "metrics" and "cvelistMetrics" (an unsupported option is echoed back under the result's warnings).	not_given

Source code in src/vulners/_resources/_sync/audit.py

def linux_audit(
    self,
    os_name: str,
    os_version: str,
    packages: Sequence[str],
    *,
    os_arch: str | None = None,
    include_unofficial: bool = False,
    include_candidates: bool = False,
    include_any_version: bool = False,
    cvelist_metrics: bool = False,
    fields: Sequence[str] | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, Any]:
    """Audit RPM/DEB/APK package lists for a Linux host.

    Each issue carries the package, its ``fixedVersion`` and ``fixedPackage``,
    and the ``applicableAdvisories`` that match it. The result also reports the
    ``appliedOptions`` that took effect and any ``warnings`` (for example an
    unsupported ``fields`` option).

    Args:
        os_name: OS name or id (``ubuntu``, ``debian``, ``rhel``, ``alpine``...).
        os_version: OS version.
        packages: Installed packages (1..2500 entries).
        os_arch: Default architecture for packages.
        include_unofficial: Include unofficial packages.
        include_candidates: Include ``candidate`` vulnerabilities.
        include_any_version: Include ``any`` version vulnerabilities.
        cvelist_metrics: Add cvelist metrics.
        fields: Advisory enrichment options to apply; this endpoint supports
            ``"metrics"`` and ``"cvelistMetrics"`` (an unsupported option is
            echoed back under the result's ``warnings``).
    """
    body: dict[str, Any] = {
        "osName": os_name,
        "osVersion": os_version,
        "packages": _require_count(
            packages, "packages", hi=_AUDIT_MAX_PACKAGES, strings=True
        ),
        "osArch": os_arch,
        "includeUnofficial": include_unofficial,
        "includeCandidates": include_candidates,
        "includeAnyVersion": include_any_version,
        "cvelistMetrics": cvelist_metrics,
    }
    self._set(body, "fields", fields, list)
    return self._request(_LINUX, body=body, timeout=timeout)

library_audit

library_audit(packages: Sequence[str], *, include_unofficial: bool = False, include_candidates: bool = False, include_any_version: bool = False, cvelist_metrics: bool = False, fields: Sequence[str] | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> dict[str, Any]

Audit a list of packages in PURL format.

Each issue carries the package, its fixedVersion and the applicableAdvisories that match it (with metrics and exploitation). The result also reports the appliedOptions that took effect and any warnings (for example an unsupported fields option).

Parameters:

Name	Type	Description	Default
packages	Sequence[str]	Packages in PURL format (1..2500 entries).	required
include_unofficial	bool	Include unofficial packages.	False
include_candidates	bool	Include candidate vulnerabilities.	False
include_any_version	bool	Include any version vulnerabilities.	False
cvelist_metrics	bool	Add cvelist metrics.	False
fields	Sequence[str] | NotGiven	Advisory enrichment options to apply; this endpoint supports "metrics" and "cvelistMetrics" (an unsupported option is echoed back under the result's warnings).	not_given

Source code in src/vulners/_resources/_sync/audit.py

def library_audit(
    self,
    packages: Sequence[str],
    *,
    include_unofficial: bool = False,
    include_candidates: bool = False,
    include_any_version: bool = False,
    cvelist_metrics: bool = False,
    fields: Sequence[str] | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, Any]:
    """Audit a list of packages in PURL format.

    Each issue carries the package, its ``fixedVersion`` and the
    ``applicableAdvisories`` that match it (with ``metrics`` and
    ``exploitation``). The result also reports the ``appliedOptions`` that took
    effect and any ``warnings`` (for example an unsupported ``fields`` option).

    Args:
        packages: Packages in PURL format (1..2500 entries).
        include_unofficial: Include unofficial packages.
        include_candidates: Include ``candidate`` vulnerabilities.
        include_any_version: Include ``any`` version vulnerabilities.
        cvelist_metrics: Add cvelist metrics.
        fields: Advisory enrichment options to apply; this endpoint supports
            ``"metrics"`` and ``"cvelistMetrics"`` (an unsupported option is
            echoed back under the result's ``warnings``).
    """
    body: dict[str, Any] = {
        "packages": _require_count(
            packages, "packages", hi=_AUDIT_MAX_PACKAGES, strings=True
        ),
        "includeUnofficial": include_unofficial,
        "includeCandidates": include_candidates,
        "includeAnyVersion": include_any_version,
        "cvelistMetrics": cvelist_metrics,
    }
    self._set(body, "fields", fields, list)
    return self._request(_LIBRARY, body=body, timeout=timeout)

sbom_audit

sbom_audit(file: str | PathLike[str], *, cvelist_metrics: bool | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> dict[str, Any]

Audit an SBOM file (SPDX or CycloneDX) for vulnerabilities.

The result reports the appliedOptions that took effect and any warnings alongside the audit data.

Parameters:

Name	Type	Description	Default
file	str | PathLike[str]	Path to the SBOM file to upload.	required
cvelist_metrics	bool | NotGiven	When True, enrich each finding with per-CVE cvelist metrics. Sent as a query option (the request body carries the uploaded file).	not_given

Source code in src/vulners/_resources/_sync/audit.py

def sbom_audit(
    self,
    file: str | os.PathLike[str],
    *,
    cvelist_metrics: bool | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, Any]:
    """Audit an SBOM file (SPDX or CycloneDX) for vulnerabilities.

    The result reports the ``appliedOptions`` that took effect and any
    ``warnings`` alongside the audit ``data``.

    Args:
        file: Path to the SBOM file to upload.
        cvelist_metrics: When ``True``, enrich each finding with per-CVE
            cvelist metrics. Sent as a query option (the request body carries
            the uploaded file).
    """
    path = os.fspath(file)
    content = vulners._base_client._call_blocking(_read_file_bytes, path)
    files = {"file": (os.path.basename(path), content, "application/json")}
    params: dict[str, Any] = {}
    self._set(params, "cvelistMetrics", cvelist_metrics)
    return self._request(_SBOM, files=files, params=params, timeout=timeout)

cve_audit

cve_audit(cve: str, *, timeout: float | Timeout | NotGiven = not_given) -> dict[str, Any]

Audit a single CVE identifier.

Parameters:

Name	Type	Description	Default
cve	str	The CVE identifier to audit, e.g. "CVE-2021-44228".	required

Source code in src/vulners/_resources/_sync/audit.py

def cve_audit(
    self,
    cve: str,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, Any]:
    """Audit a single CVE identifier.

    Args:
        cve: The CVE identifier to audit, e.g. ``"CVE-2021-44228"``.
    """
    return self._request(_CVE, body={"cve": cve}, timeout=timeout)

cve_batch_audit

cve_batch_audit(cve: Sequence[str], *, timeout: float | Timeout | NotGiven = not_given) -> list[dict[str, Any]]

Audit a batch of CVE identifiers.

Parameters:

Name	Type	Description	Default
cve	Sequence[str]	CVE identifiers to audit; at least one, e.g. ["CVE-2021-44228", "CVE-2017-0144"].	required

Returns:

Type	Description
list[dict[str, Any]]	A list of audit results for the submitted CVEs.

Source code in src/vulners/_resources/_sync/audit.py

def cve_batch_audit(
    self,
    cve: Sequence[str],
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> list[dict[str, Any]]:
    """Audit a batch of CVE identifiers.

    Args:
        cve: CVE identifiers to audit; at least one, e.g.
            ``["CVE-2021-44228", "CVE-2017-0144"]``.

    Returns:
        A list of audit results for the submitted CVEs.
    """
    return self._request(
        _CVES, body={"cve": _require_count(cve, "cve", strings=True)}, timeout=timeout
    )

kb_audit

kb_audit(os_name: str, kb_list: Sequence[str], *, os_version: str | NotGiven = not_given, fields: Sequence[str] | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> dict[str, Any]

Audit a Windows host for missing security updates (/api/v4/audit/kb).

Reports one finding per missing update: the result's items list holds a package with the fixedPackage (the KB that fixes it) and the advisories for that update — each advisory carrying the KBs it supersedes alongside its severity, family and affected products. This replaces the flat CVE list of the deprecated v3 endpoint (:meth:kb_audit_v3).

Parameters:

Name	Type	Description	Default
os_name	str	Windows OS name, e.g. "Windows 10" / "Windows Server 2012 R2".	required
kb_list	Sequence[str]	Installed KBs, e.g. ["KB5028166"].	required
os_version	str | NotGiven	Optional OS build (e.g. "10.0.19045") to disambiguate the edition.	not_given
fields	Sequence[str] | NotGiven	Advisory fields to include; server default when omitted.	not_given

Returns:

Type	Description
dict[str, Any]	A result mapping with items (one entry per package, each with its
dict[str, Any]	fixedPackage and advisories) and totalPackages.

Source code in src/vulners/_resources/_sync/audit.py

def kb_audit(
    self,
    os_name: str,
    kb_list: Sequence[str],
    *,
    os_version: str | NotGiven = not_given,
    fields: Sequence[str] | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, Any]:
    """Audit a Windows host for missing security updates (``/api/v4/audit/kb``).

    Reports one finding per missing update: the result's ``items`` list holds a
    package with the ``fixedPackage`` (the KB that fixes it) and the
    ``advisories`` for that update — each advisory carrying the KBs it
    ``supersedes`` alongside its severity, family and affected products. This
    replaces the flat CVE list of the deprecated v3 endpoint
    (:meth:`kb_audit_v3`).

    Args:
        os_name: Windows OS name, e.g. ``"Windows 10"`` /
            ``"Windows Server 2012 R2"``.
        kb_list: Installed KBs, e.g. ``["KB5028166"]``.
        os_version: Optional OS build (e.g. ``"10.0.19045"``) to disambiguate
            the edition.
        fields: Advisory fields to include; server default when omitted.

    Returns:
        A result mapping with ``items`` (one entry per package, each with its
        ``fixedPackage`` and ``advisories``) and ``totalPackages``.
    """
    body: dict[str, Any] = {"osName": os_name, "kbList": list(kb_list)}
    self._set(body, "osVersion", os_version)
    self._set(body, "fields", fields, list)
    return self._request(_KB_V4, body=body, timeout=timeout)

kb_audit_v3

kb_audit_v3(os: str, kb_list: Sequence[str], *, timeout: float | Timeout | NotGiven = not_given) -> dict[str, Any]

Audit a Windows host by its installed KBs (deprecated v3 endpoint).

.. deprecated:: Uses the legacy /api/v3/audit/kb/ endpoint, which returns a flat result (kbLatest, kbMissed, cvelist). Prefer :meth:kb_audit, which uses /api/v4/audit/kb and returns richer per-update advisories.

Parameters:

Name	Type	Description	Default
os	str	Windows OS name, e.g. "Windows Server 2012 R2".	required
kb_list	Sequence[str]	Installed KBs, e.g. ["KB2918614", "KB2918616"].	required

Source code in src/vulners/_resources/_sync/audit.py

def kb_audit_v3(
    self,
    os: str,
    kb_list: Sequence[str],
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, Any]:
    """Audit a Windows host by its installed KBs (deprecated v3 endpoint).

    .. deprecated::
        Uses the legacy ``/api/v3/audit/kb/`` endpoint, which returns a flat
        result (``kbLatest``, ``kbMissed``, ``cvelist``). Prefer :meth:`kb_audit`,
        which uses ``/api/v4/audit/kb`` and returns richer per-update advisories.

    Args:
        os: Windows OS name, e.g. ``"Windows Server 2012 R2"``.
        kb_list: Installed KBs, e.g. ``["KB2918614", "KB2918616"]``.
    """
    body = {"os": os, "kbList": list(kb_list)}
    return self._request(_KB_V3, body=body, timeout=timeout)

win_audit

win_audit(os: str, os_version: str, kb_list: Sequence[str], software: Sequence[WinAuditItem], *, platform: str | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> dict[str, Any]

Audit a Windows host by installed KBs and software.

Parameters:

Name	Type	Description	Default
os	str	Windows OS name, e.g. "Windows Server 2012 R2".	required
os_version	str	Windows OS version, e.g. "10.0.19045".	required
kb_list	Sequence[str]	Installed KBs.	required
software	Sequence[WinAuditItem]	Installed software, {"software": ..., "version": ...}.	required
platform	str | NotGiven	OS platform, e.g. "x86".	not_given

Source code in src/vulners/_resources/_sync/audit.py

def win_audit(
    self,
    os: str,
    os_version: str,
    kb_list: Sequence[str],
    software: Sequence[WinAuditItem],
    *,
    platform: str | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, Any]:
    """Audit a Windows host by installed KBs and software.

    Args:
        os: Windows OS name, e.g. ``"Windows Server 2012 R2"``.
        os_version: Windows OS version, e.g. ``"10.0.19045"``.
        kb_list: Installed KBs.
        software: Installed software, ``{"software": ..., "version": ...}``.
        platform: OS platform, e.g. ``"x86"``.
    """
    body: dict[str, Any] = {
        "os": os,
        "os_version": os_version,
        "kb_list": list(kb_list),
        "software": list(software),
    }
    self._set(body, "platform", platform)
    # This endpoint requires the api key echoed in the request body.
    body["apiKey"] = self._api_key
    return self._request(_WINAUDIT, body=body, timeout=timeout)

smart

smart(software: Sequence[str], *, catalog: Literal['official', 'extended'] = 'official', fields: Sequence[str] | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> list[dict[str, Any]]

Resolve free-form software strings to CPE/PURLs and their vulnerabilities.

Each input string is matched heuristically to a CPE and/or PURLs, returning {"input", "cpe"?, "purls", "confidence", "fixedVersion", "vulnerabilities"} per entry.

Note

This is a preview endpoint. Billing is per submitted string (every entry in software is charged), so keep the batch to what you need.

Parameters:

Name	Type	Description	Default
software	Sequence[str]	1..500 strings, each at most 512 characters.	required
catalog	Literal['official', 'extended']	"official" or "extended" product catalog.	'official'
fields	Sequence[str] | NotGiven	Fields to include on each vulnerability (a projection). Valid values include "metrics", "exploitation", "cvelist", "cvelistMetrics", "epss", "description" and the other bulletin fields; the server rejects an unknown value with 400. Server default projection when omitted.	not_given

Raises:

Type	Description
ValueError	software is empty, has more than 500 entries, or any entry exceeds 512 characters.

Source code in src/vulners/_resources/_sync/audit.py

def smart(
    self,
    software: Sequence[str],
    *,
    catalog: Literal["official", "extended"] = "official",
    fields: Sequence[str] | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> list[dict[str, Any]]:
    """Resolve free-form software strings to CPE/PURLs and their vulnerabilities.

    Each input string is matched heuristically to a CPE and/or PURLs, returning
    ``{"input", "cpe"?, "purls", "confidence", "fixedVersion", "vulnerabilities"}``
    per entry.

    Note:
        This is a preview endpoint. **Billing is per submitted string** (every
        entry in ``software`` is charged), so keep the batch to what you need.

    Args:
        software: 1..500 strings, each at most 512 characters.
        catalog: ``"official"`` or ``"extended"`` product catalog.
        fields: Fields to include on each vulnerability (a projection). Valid
            values include ``"metrics"``, ``"exploitation"``, ``"cvelist"``,
            ``"cvelistMetrics"``, ``"epss"``, ``"description"`` and the other
            bulletin fields; the server rejects an unknown value with ``400``.
            Server default projection when omitted.

    Raises:
        ValueError: ``software`` is empty, has more than 500 entries, or any
            entry exceeds 512 characters.
    """
    items = list(software)
    if not 1 <= len(items) <= _SMART_MAX_ITEMS:
        raise ValueError(
            f"software must have between 1 and {_SMART_MAX_ITEMS} entries "
            f"(got {len(items)}); billing is per submitted string."
        )
    for item in items:
        if len(item) > _SMART_MAX_LEN:
            raise ValueError(
                f"each software entry must be at most {_SMART_MAX_LEN} characters"
            )
    body: dict[str, Any] = {"software": items, "catalog": catalog}
    self._set(body, "fields", fields, list)
    return self._request(_SMART, body=body, timeout=timeout)

metadata

metadata(registry: str, name: str, version: str, *, timeout: float | Timeout | NotGiven = not_given) -> PackageMetadata

Look up a single package's license and version-range metadata.

Returns the package's declared licenses (as a list) along with the version range the metadata covers. The endpoint is public and available on all plans.

Two inputs are normalized for you: registry is lower-cased, and for Maven the name is the groupId:artifactId coordinate — a "/" is converted to the required ":".

Distinguishing outcomes (see :class:~vulners.PackageMetadata):

• Known package with licenses — :attr:~vulners.PackageMetadata.license is a non-empty list.
• Known package, no recorded license — :attr:~vulners.PackageMetadata.found is True but license is [].
• Package unknown to the registry — :attr:~vulners.PackageMetadata.found is False (the endpoint returns an empty range).
• API / network error — a :class:~vulners.VulnersError is raised; an empty license list is never a silent stand-in for an error.

Parameters:

Name	Type	Description	Default
registry	str	Package registry, e.g. "pypi", "npm", "maven" (case is normalized).	required
name	str	Package name. For Maven, "groupId:artifactId" (a "/" is converted to ":").	required
version	str	Package version, e.g. "2.28.0".	required

Returns:

Name	Type	Description
A	PackageMetadata	class:~vulners.PackageMetadata with name, version,
	PackageMetadata	range and license (a list).

Source code in src/vulners/_resources/_sync/audit.py

def metadata(
    self,
    registry: str,
    name: str,
    version: str,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> PackageMetadata:
    """Look up a single package's license and version-range metadata.

    Returns the package's declared licenses (as a list) along with the version
    ``range`` the metadata covers. The endpoint is public and available on all
    plans.

    Two inputs are normalized for you: ``registry`` is lower-cased, and for
    Maven the ``name`` is the ``groupId:artifactId`` coordinate — a ``"/"`` is
    converted to the required ``":"``.

    Distinguishing outcomes (see :class:`~vulners.PackageMetadata`):

    - **Known package with licenses** — :attr:`~vulners.PackageMetadata.license`
      is a non-empty list.
    - **Known package, no recorded license** —
      :attr:`~vulners.PackageMetadata.found` is ``True`` but ``license`` is ``[]``.
    - **Package unknown to the registry** —
      :attr:`~vulners.PackageMetadata.found` is ``False`` (the endpoint returns
      an empty ``range``).
    - **API / network error** — a :class:`~vulners.VulnersError` is raised; an
      empty license list is never a silent stand-in for an error.

    Args:
        registry: Package registry, e.g. ``"pypi"``, ``"npm"``, ``"maven"``
            (case is normalized).
        name: Package name. For Maven, ``"groupId:artifactId"`` (a ``"/"`` is
            converted to ``":"``).
        version: Package version, e.g. ``"2.28.0"``.

    Returns:
        A :class:`~vulners.PackageMetadata` with ``name``, ``version``,
        ``range`` and ``license`` (a list).
    """
    reg = registry.lower()
    pkg = name.replace("/", ":") if reg == "maven" else name
    body = {"registry": reg, "name": pkg, "version": version}
    return self._request(_METADATA, body=body, cast=_package_metadata, timeout=timeout)

Archive

vulners._resources._sync.archive.Archive

Archive(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

Download bulk archives of the Vulners database.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

fetch_collection

fetch_collection(type: str, *, timeout: float | Timeout | NotGiven = not_given) -> Any

Download an entire collection archive by type (e.g. "cve").

Buffers and decodes the whole archive in memory. For large collections prefer :meth:iter_collection (lazy, per-element) or :meth:download_collection (parallel, straight to disk).

Parameters:

Name	Type	Description	Default
type	str	The collection type to download (e.g. "cve").	required

Returns:

Type	Description
Any	The decoded collection — parsed JSON (typically a list of records)
Any	when the body decodes, otherwise the raw archive bytes.

Source code in src/vulners/_resources/_sync/archive.py

def fetch_collection(
    self,
    type: str,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Download an entire collection archive by ``type`` (e.g. ``"cve"``).

    Buffers and decodes the whole archive in memory. For large collections
    prefer :meth:`iter_collection` (lazy, per-element) or
    :meth:`download_collection` (parallel, straight to disk).

    Args:
        type: The collection type to download (e.g. ``"cve"``).

    Returns:
        The decoded collection — parsed JSON (typically a list of records)
        when the body decodes, otherwise the raw archive bytes.
    """
    return self._request(
        _FETCH_COLLECTION, cast=_decode_archive, body={"type": type}, timeout=timeout
    )

iter_collection

iter_collection(type: str, *, timeout: float | Timeout | NotGiven = not_given) -> Iterator[dict[str, Any]]

Stream a collection archive element by element (a JSON array).

Unlike :meth:fetch_collection (which buffers and decodes the whole archive), this follows the archive redirect to storage, decompresses the body as a stream and yields each array element lazily, so a multi-gigabyte collection never has to be held in memory. Records delivered as raw Elasticsearch hits are normalized to their "_source" document.

Parameters:

Name	Type	Description	Default
type	str	The collection type to stream (e.g. "cve").	required

Yields:

Type	Description
dict[str, Any]	Each collection element as a dict.

Source code in src/vulners/_resources/_sync/archive.py

def iter_collection(
    self,
    type: str,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Iterator[dict[str, Any]]:
    """Stream a collection archive element by element (a JSON array).

    Unlike :meth:`fetch_collection` (which buffers and decodes the whole
    archive), this follows the archive redirect to storage, decompresses the
    body as a stream and yields each array element lazily, so a multi-gigabyte
    collection never has to be held in memory. Records delivered as raw
    Elasticsearch hits are normalized to their ``"_source"`` document.

    Args:
        type: The collection type to stream (e.g. ``"cve"``).

    Yields:
        Each collection element as a ``dict``.
    """
    for record in self._client.stream_records(
        _STREAM_COLLECTION, params={"type": type}, timeout=timeout
    ):
        yield record

download_collection

download_collection(collection: str, path: str | PathLike[str], *, update_from: datetime | None = None, connections: int = 8, timeout: float | Timeout | NotGiven = not_given) -> int

Download a collection archive to path, in parallel; return bytes written.

The endpoint redirects to storage that supports HTTP range requests, so the raw (still-compressed) archive is pulled over connections concurrent connections and written straight to disk — saturating the link in constant memory — with an automatic fallback to a single stream when the storage does not offer ranges. Nothing is decompressed, so a multi-gigabyte collection downloads without ever being held in memory. Pass update_from to fetch only the entries changed after that moment (the collection-update endpoint). The write is atomic: an interrupted download never clobbers an existing file.

Parameters:

Name	Type	Description	Default
collection	str	The collection type to download (e.g. "cve").	required
path	str | PathLike[str]	Destination file path; an existing file is overwritten atomically.	required
update_from	datetime | None	When given, download the collection update since this moment instead of the full archive.	None
connections	int	Number of parallel range connections (default 8). The SDK-owned client runs them over HTTP/1.1 so each opens a real socket and saturates the link; with your own http_client pass http2=False for the same throughput.	8

Returns:

Type	Description
int	The number of bytes written to path.

Source code in src/vulners/_resources/_sync/archive.py

def download_collection(
    self,
    collection: str,
    path: str | os.PathLike[str],
    *,
    update_from: datetime | None = None,
    connections: int = 8,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> int:
    """Download a collection archive to ``path``, in parallel; return bytes written.

    The endpoint redirects to storage that supports HTTP range requests, so the
    raw (still-compressed) archive is pulled over ``connections`` concurrent
    connections and written straight to disk — saturating the link in constant
    memory — with an automatic fallback to a single stream when the storage does
    not offer ranges. Nothing is decompressed, so a multi-gigabyte collection
    downloads without ever being held in memory. Pass ``update_from`` to fetch
    only the entries changed after that moment (the collection-update endpoint).
    The write is atomic: an interrupted download never clobbers an existing file.

    Args:
        collection: The collection type to download (e.g. ``"cve"``).
        path: Destination file path; an existing file is overwritten atomically.
        update_from: When given, download the collection update since this
            moment instead of the full archive.
        connections: Number of parallel range connections (default 8). The
            SDK-owned client runs them over HTTP/1.1 so each opens a real socket
            and saturates the link; with your own ``http_client`` pass
            ``http2=False`` for the same throughput.

    Returns:
        The number of bytes written to ``path``.
    """
    if connections < 1:
        raise ValueError("connections must be >= 1")
    spec = _STREAM_COLLECTION
    params: dict[str, Any] = {"type": collection}
    if update_from is not None:
        spec = _STREAM_COLLECTION_UPDATE
        params["after"] = update_from.isoformat()
    return _download.parallel_download_sync(
        self._client,
        spec,
        os.fspath(path),
        params=params,
        connections=connections,
        timeout=timeout,
    )

fetch_collection_update

fetch_collection_update(type: str, after: datetime, *, timeout: float | Timeout | NotGiven = not_given) -> Any

Download only the collection entries changed after after.

The incremental counterpart of :meth:fetch_collection, buffered and decoded in memory. Use :meth:collection_state to obtain the cursor to resume from.

Parameters:

Name	Type	Description	Default
type	str	The collection type to download (e.g. "cve").	required
after	datetime	Only entries changed after this moment are included.	required

Returns:

Type	Description
Any	The decoded update — parsed JSON (typically a list of records) when
Any	the body decodes, otherwise the raw archive bytes.

Source code in src/vulners/_resources/_sync/archive.py

def fetch_collection_update(
    self,
    type: str,
    after: datetime,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Download only the collection entries changed after ``after``.

    The incremental counterpart of :meth:`fetch_collection`, buffered and
    decoded in memory. Use :meth:`collection_state` to obtain the cursor to
    resume from.

    Args:
        type: The collection type to download (e.g. ``"cve"``).
        after: Only entries changed after this moment are included.

    Returns:
        The decoded update — parsed JSON (typically a list of records) when
        the body decodes, otherwise the raw archive bytes.
    """
    body = {"type": type, "after": after.isoformat()}
    return self._request(
        _FETCH_COLLECTION_UPDATE, cast=_decode_archive, body=body, timeout=timeout
    )

collection_state

collection_state(type: str, *, timeout: float | Timeout | NotGiven = not_given) -> Any

Read the sync cursor and counters for a collection.

Returns:

Type	Description
Any	A dict with cursor (feed it back as after to the
Any	collection-update download), upload_time, write_time and
Any	total_docs.

Source code in src/vulners/_resources/_sync/archive.py

def collection_state(
    self,
    type: str,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Read the sync cursor and counters for a collection.

    Returns:
        A dict with ``cursor`` (feed it back as ``after`` to the
        collection-update download), ``upload_time``, ``write_time`` and
        ``total_docs``.
    """
    return self._request(_COLLECTION_STATE, body={"type": type}, timeout=timeout)

family

family(name: str, *, timeout: float | Timeout | NotGiven = not_given) -> Any

Download an entire collection-family archive by name.

Same shape as :meth:fetch_collection, keyed by a family name (e.g. "exploit", "unix", "software") instead of a single collection type. Buffered and decoded in memory; for large families prefer :meth:iter_family (lazy, per-element).

Parameters:

Name	Type	Description	Default
name	str	The collection family to download (e.g. "exploit").	required

Returns:

Type	Description
Any	The decoded family archive — parsed JSON (typically a list of
Any	records) when the body decodes, otherwise the raw archive bytes.

Source code in src/vulners/_resources/_sync/archive.py

def family(
    self,
    name: str,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Download an entire collection-family archive by ``name``.

    Same shape as :meth:`fetch_collection`, keyed by a family name (e.g.
    ``"exploit"``, ``"unix"``, ``"software"``) instead of a single
    collection type. Buffered and decoded in memory; for large families
    prefer :meth:`iter_family` (lazy, per-element).

    Args:
        name: The collection family to download (e.g. ``"exploit"``).

    Returns:
        The decoded family archive — parsed JSON (typically a list of
        records) when the body decodes, otherwise the raw archive bytes.
    """
    return self._request(
        _FETCH_FAMILY, cast=_decode_archive, body={"name": name}, timeout=timeout
    )

family_update

family_update(name: str, after: datetime, *, timeout: float | Timeout | NotGiven = not_given) -> Any

Download only the family entries changed after after (max 25h ago).

The incremental counterpart of :meth:family. Use :meth:family_state to obtain the cursor to resume from.

Parameters:

Name	Type	Description	Default
name	str	The collection family to download (e.g. "exploit").	required
after	datetime	Only entries changed after this moment are included; must be at most 25 hours ago.	required

Returns:

Type	Description
Any	The decoded update — parsed JSON (typically a list of records) when
Any	the body decodes, otherwise the raw archive bytes.

Source code in src/vulners/_resources/_sync/archive.py

def family_update(
    self,
    name: str,
    after: datetime,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Download only the family entries changed after ``after`` (max 25h ago).

    The incremental counterpart of :meth:`family`. Use :meth:`family_state`
    to obtain the cursor to resume from.

    Args:
        name: The collection family to download (e.g. ``"exploit"``).
        after: Only entries changed after this moment are included; must be
            at most 25 hours ago.

    Returns:
        The decoded update — parsed JSON (typically a list of records) when
        the body decodes, otherwise the raw archive bytes.
    """
    body = {"name": name, "after": after.isoformat()}
    return self._request(
        _FETCH_FAMILY_UPDATE, cast=_decode_archive, body=body, timeout=timeout
    )

family_state

family_state(name: str, *, timeout: float | Timeout | NotGiven = not_given) -> Any

Read the sync cursor and counters for a collection family.

Returns:

Type	Description
Any	A dict with cursor (feed it back as after to
Any	meth:family_update), upload_time, write_time and
Any	total_docs.

Source code in src/vulners/_resources/_sync/archive.py

def family_state(
    self,
    name: str,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Read the sync cursor and counters for a collection family.

    Returns:
        A dict with ``cursor`` (feed it back as ``after`` to
        :meth:`family_update`), ``upload_time``, ``write_time`` and
        ``total_docs``.
    """
    return self._request(_FAMILY_STATE, body={"name": name}, timeout=timeout)

iter_family

iter_family(name: str, *, update_from: datetime | None = None, timeout: float | Timeout | NotGiven = not_given) -> Iterator[dict[str, Any]]

Stream a family archive element by element, like :meth:iter_collection.

Follows the archive redirect to storage, decompresses the body as a stream and yields each element lazily, so a multi-gigabyte family archive never has to be held in memory.

Parameters:

Name	Type	Description	Default
name	str	The collection family to stream (e.g. "exploit").	required
update_from	datetime | None	When given, stream only the entries changed after this moment (at most 25 hours ago) instead of the whole family.	None

Yields:

Type	Description
dict[str, Any]	Each family element as a dict.

Source code in src/vulners/_resources/_sync/archive.py

def iter_family(
    self,
    name: str,
    *,
    update_from: datetime | None = None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Iterator[dict[str, Any]]:
    """Stream a family archive element by element, like :meth:`iter_collection`.

    Follows the archive redirect to storage, decompresses the body as a
    stream and yields each element lazily, so a multi-gigabyte family
    archive never has to be held in memory.

    Args:
        name: The collection family to stream (e.g. ``"exploit"``).
        update_from: When given, stream only the entries changed after this
            moment (at most 25 hours ago) instead of the whole family.

    Yields:
        Each family element as a ``dict``.
    """
    spec = _STREAM_FAMILY
    params: dict[str, Any] = {"name": name}
    if update_from is not None:
        spec = _STREAM_FAMILY_UPDATE
        params["after"] = update_from.isoformat()
    for record in self._client.stream_records(spec, params=params, timeout=timeout):
        yield record

get_collection

get_collection(type: str, *, datefrom: str = '1976-01-01', dateto: str = '2199-01-01', timeout: float | Timeout | NotGiven = not_given) -> Any

Download a collection over a date range (legacy v3 endpoint).

Buffered and decoded in memory. Prefer the v4 :meth:fetch_collection / :meth:download_collection where available.

Parameters:

Name	Type	Description	Default
type	str	The collection type to download (e.g. "cve").	required
datefrom	str	Start date (YYYY-MM-DD) of the range.	'1976-01-01'
dateto	str	End date (YYYY-MM-DD) of the range.	'2199-01-01'

Returns:

Type	Description
Any	The decoded collection — parsed JSON (typically a list of records)
Any	when the body decodes, otherwise the raw archive bytes.

Source code in src/vulners/_resources/_sync/archive.py

def get_collection(
    self,
    type: str,
    *,
    datefrom: str = "1976-01-01",
    dateto: str = "2199-01-01",
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Download a collection over a date range (legacy v3 endpoint).

    Buffered and decoded in memory. Prefer the v4 :meth:`fetch_collection` /
    :meth:`download_collection` where available.

    Args:
        type: The collection type to download (e.g. ``"cve"``).
        datefrom: Start date (``YYYY-MM-DD``) of the range.
        dateto: End date (``YYYY-MM-DD``) of the range.

    Returns:
        The decoded collection — parsed JSON (typically a list of records)
        when the body decodes, otherwise the raw archive bytes.
    """
    body = {"type": type, "datefrom": datefrom, "dateto": dateto}
    return self._request(_GET_COLLECTION, cast=_decode_archive, body=body, timeout=timeout)

get_distributive

get_distributive(os: str, version: str, *, timeout: float | Timeout | NotGiven = not_given) -> list[Any]

Download the vulnerability distributive for an OS/version (legacy v3).

Parameters:

Name	Type	Description	Default
os	str	The operating system identifier (e.g. "debian", "centos").	required
version	str	The OS version (e.g. "11").	required

Returns:

Type	Description
list[Any]	A list of the distributive's vulnerability documents (each the
list[Any]	"_source" of a record).

Source code in src/vulners/_resources/_sync/archive.py

def get_distributive(
    self,
    os: str,
    version: str,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> list[Any]:
    """Download the vulnerability distributive for an OS/version (legacy v3).

    Args:
        os: The operating system identifier (e.g. ``"debian"``, ``"centos"``).
        version: The OS version (e.g. ``"11"``).

    Returns:
        A list of the distributive's vulnerability documents (each the
        ``"_source"`` of a record).
    """
    body = {"os": os, "version": version}
    return self._request(_GET_DISTRIBUTIVE, cast=_distributive, body=body, timeout=timeout)

getsploit

getsploit(*, timeout: float | Timeout | NotGiven = not_given) -> bytes

Download the raw getsploit exploit database archive (legacy v3).

Buffers the whole archive in memory; for the full database prefer :meth:download_getsploit, which streams it to disk in parallel.

Returns:

Type	Description
bytes	The raw archive bytes (a single-member zip whose member is the
bytes	getsploit SQLite database).

Source code in src/vulners/_resources/_sync/archive.py

def getsploit(
    self,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> bytes:
    """Download the raw getsploit exploit database archive (legacy v3).

    Buffers the whole archive in memory; for the full database prefer
    :meth:`download_getsploit`, which streams it to disk in parallel.

    Returns:
        The raw archive bytes (a single-member zip whose member is the
        getsploit SQLite database).
    """
    return self._request(_GETSPLOIT, timeout=timeout)

download_getsploit

download_getsploit(path: str | PathLike[str], *, connections: int = 8, timeout: float | Timeout | NotGiven = not_given) -> int

Stream the getsploit database archive to path, in parallel; return bytes written.

The endpoint redirects to storage that supports HTTP range requests, so the archive is pulled over connections concurrent connections and written straight to disk — saturating the link and using constant memory — instead of buffering the whole database in RAM like :meth:getsploit. Prefer this for the full database; use :meth:getsploit only for the raw bytes in memory. The write is atomic (an interrupted download never clobbers an existing file), and the tool falls back to a single stream if the storage does not offer ranges.

The written file is the raw archive (a single-member zip whose member is the getsploit SQLite database); unzip it to obtain getsploit.db.

Legacy v3 endpoint; there is no v4 equivalent and the server may retire it during the 4.x lifetime.

Parameters:

Name	Type	Description	Default
path	str | PathLike[str]	Destination file path; an existing file is overwritten atomically.	required
connections	int	Number of parallel range connections (default 8). The SDK-owned client runs them over HTTP/1.1 so each opens a real socket and saturates the link; with your own http_client pass http2=False for the same throughput.	8

Returns:

Type	Description
int	The number of bytes written to path.

Source code in src/vulners/_resources/_sync/archive.py

def download_getsploit(
    self,
    path: str | os.PathLike[str],
    *,
    connections: int = 8,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> int:
    """Stream the getsploit database archive to ``path``, in parallel; return bytes written.

    The endpoint redirects to storage that supports HTTP range requests, so the
    archive is pulled over ``connections`` concurrent connections and written
    straight to disk — saturating the link and using constant memory — instead
    of buffering the whole database in RAM like :meth:`getsploit`. Prefer this
    for the full database; use :meth:`getsploit` only for the raw bytes in memory.
    The write is atomic (an interrupted download never clobbers an existing file),
    and the tool falls back to a single stream if the storage does not offer ranges.

    The written file is the raw archive (a single-member zip whose member is the
    getsploit SQLite database); unzip it to obtain ``getsploit.db``.

    Legacy v3 endpoint; there is no v4 equivalent and the server may retire it
    during the 4.x lifetime.

    Args:
        path: Destination file path; an existing file is overwritten atomically.
        connections: Number of parallel range connections (default 8). The
            SDK-owned client runs them over HTTP/1.1 so each opens a real socket
            and saturates the link; with your own ``http_client`` pass
            ``http2=False`` for the same throughput.

    Returns:
        The number of bytes written to ``path``.
    """
    if connections < 1:
        raise ValueError("connections must be >= 1")
    return _download.parallel_download_sync(
        self._client,
        _STREAM_GETSPLOIT,
        os.fspath(path),
        connections=connections,
        timeout=timeout,
    )

Misc

vulners._resources._sync.misc.Misc

Misc(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

Miscellaneous search and metadata helpers.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

search_cpe

search_cpe(product: str, *, vendor: str | NotGiven = not_given, size: int | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> Any

Search for CPE strings matching a product (and optional vendor).

Parameters:

Name	Type	Description	Default
product	str	Product string to search a CPE for.	required
vendor	str | NotGiven	Optional vendor to narrow the match.	not_given
size	int | NotGiven	Maximum number of results (0..10000, 0 means all).	not_given

Returns:

Type	Description
Any	The CPE search result for product. Also reachable as
Any	meth:Search.cpe.

Source code in src/vulners/_resources/_sync/misc.py

def search_cpe(
    self,
    product: str,
    *,
    vendor: str | NotGiven = not_given,
    size: int | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Search for CPE strings matching a product (and optional vendor).

    Args:
        product: Product string to search a CPE for.
        vendor: Optional vendor to narrow the match.
        size: Maximum number of results (0..10000, ``0`` means all).

    Returns:
        The CPE search result for ``product``. Also reachable as
        :meth:`Search.cpe`.
    """
    body: dict[str, Any] = {"product": product}
    self._set(body, "vendor", vendor)
    self._set(body, "size", size)
    return self._request(_SEARCH_CPE, body=body, timeout=timeout)

query_autocomplete

query_autocomplete(query: str, *, timeout: float | Timeout | NotGiven = not_given) -> list[str | list[str]]

Return possible completions for a partial Lucene query.

Most suggestions are strings; the server occasionally returns a group of related completions, which arrives as a list[str] element.

Parameters:

Name	Type	Description	Default
query	str	The partial Lucene query to complete.	required

Returns:

Type	Description
list[str | list[str]]	The completions, each a single string or a list[str] of related
list[str | list[str]]	completions. Also reachable as :meth:Search.autocomplete.

Source code in src/vulners/_resources/_sync/misc.py

def query_autocomplete(
    self,
    query: str,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> list[str | list[str]]:
    """Return possible completions for a partial Lucene query.

    Most suggestions are strings; the server occasionally returns a group of
    related completions, which arrives as a ``list[str]`` element.

    Args:
        query: The partial Lucene query to complete.

    Returns:
        The completions, each a single string or a ``list[str]`` of related
        completions. Also reachable as :meth:`Search.autocomplete`.
    """
    return self._request(
        _AUTOCOMPLETE, cast=_autocomplete, body={"query": query}, timeout=timeout
    )

get_suggestion

get_suggestion(field_name: str, *, type: Literal['distinct'] = 'distinct', timeout: float | Timeout | NotGiven = not_given) -> Any

Return distinct value suggestions for a document field.

Parameters:

Name	Type	Description	Default
field_name	str	The document field to suggest values for.	required
type	Literal['distinct']	Suggestion type; only "distinct" is supported.	'distinct'

Returns:

Type	Description
Any	The distinct values observed for field_name. Also reachable as
Any	meth:Search.suggest.

Source code in src/vulners/_resources/_sync/misc.py

def get_suggestion(
    self,
    field_name: str,
    *,
    type: Literal["distinct"] = "distinct",
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Return distinct value suggestions for a document field.

    Args:
        field_name: The document field to suggest values for.
        type: Suggestion type; only ``"distinct"`` is supported.

    Returns:
        The distinct values observed for ``field_name``. Also reachable as
        :meth:`Search.suggest`.
    """
    body = {"fieldName": field_name, "type": type}
    return self._request(_SUGGEST, body=body, timeout=timeout)

get_web_application_rules

get_web_application_rules(*, timeout: float | Timeout | NotGiven = not_given) -> Any

Return the Vulners web-application (burp) detection rule set.

Returns:

Type	Description
Any	The web-application (burp) detection rules. Also reachable as
Any	meth:Search.web_vulns.

Source code in src/vulners/_resources/_sync/misc.py

def get_web_application_rules(
    self,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Return the Vulners web-application (burp) detection rule set.

    Returns:
        The web-application (burp) detection rules. Also reachable as
        :meth:`Search.web_vulns`.
    """
    return self._request(_BURP_RULES, timeout=timeout)

Report

vulners._resources._sync.report.Report

Report(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

Reports over Linux-audit results.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

vulns_summary

vulns_summary(*, limit: int = 30, offset: int = 0, filter: dict[str, Any] | None = None, sort: str = '', timeout: float | Timeout | NotGiven = not_given) -> Any

Summarise every found vulnerability (id, title, score, severity...).

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of rows to return in this page.	30
offset	int	Number of rows to skip.	0
filter	dict[str, Any] | None	Additional report filter, if any.	None
sort	str	Sort field; prefix with - for descending.	''

Returns:

Type	Description
Any	The vulnerability-summary report payload: one row per distinct vulnerability.

Source code in src/vulners/_resources/_sync/report.py

def vulns_summary(
    self,
    *,
    limit: int = 30,
    offset: int = 0,
    filter: dict[str, Any] | None = None,
    sort: str = "",
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Summarise every found vulnerability (id, title, score, severity...).

    Args:
        limit: Maximum number of rows to return in this page.
        offset: Number of rows to skip.
        filter: Additional report filter, if any.
        sort: Sort field; prefix with ``-`` for descending.

    Returns:
        The vulnerability-summary report payload: one row per distinct vulnerability.
    """
    return self._report("vulnssummary", limit, offset, filter, sort, timeout)

vulns_list

vulns_list(*, limit: int = 30, offset: int = 0, filter: dict[str, Any] | None = None, sort: str = '', timeout: float | Timeout | NotGiven = not_given) -> Any

List vulnerabilities found on hosts, with host information.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of rows to return in this page.	30
offset	int	Number of rows to skip.	0
filter	dict[str, Any] | None	Additional report filter, if any.	None
sort	str	Sort field; prefix with - for descending.	''

Returns:

Type	Description
Any	The vulnerability-list report payload: one row per vulnerability occurrence on a host.

Source code in src/vulners/_resources/_sync/report.py

def vulns_list(
    self,
    *,
    limit: int = 30,
    offset: int = 0,
    filter: dict[str, Any] | None = None,
    sort: str = "",
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """List vulnerabilities found on hosts, with host information.

    Args:
        limit: Maximum number of rows to return in this page.
        offset: Number of rows to skip.
        filter: Additional report filter, if any.
        sort: Sort field; prefix with ``-`` for descending.

    Returns:
        The vulnerability-list report payload: one row per vulnerability occurrence on a host.
    """
    return self._report("vulnslist", limit, offset, filter, sort, timeout)

ip_summary

ip_summary(*, limit: int = 30, offset: int = 0, filter: dict[str, Any] | None = None, sort: str = '', timeout: float | Timeout | NotGiven = not_given) -> Any

Summarise results per host (agent id, ip, fqdn, os, vulnerability counts).

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of rows to return in this page.	30
offset	int	Number of rows to skip.	0
filter	dict[str, Any] | None	Additional report filter, if any.	None
sort	str	Sort field; prefix with - for descending.	''

Returns:

Type	Description
Any	The per-host summary report payload: one row per host.

Source code in src/vulners/_resources/_sync/report.py

def ip_summary(
    self,
    *,
    limit: int = 30,
    offset: int = 0,
    filter: dict[str, Any] | None = None,
    sort: str = "",
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Summarise results per host (agent id, ip, fqdn, os, vulnerability counts).

    Args:
        limit: Maximum number of rows to return in this page.
        offset: Number of rows to skip.
        filter: Additional report filter, if any.
        sort: Sort field; prefix with ``-`` for descending.

    Returns:
        The per-host summary report payload: one row per host.
    """
    return self._report("ipsummary", limit, offset, filter, sort, timeout)

scan_list

scan_list(*, limit: int = 30, offset: int = 0, filter: dict[str, Any] | None = None, sort: str = '', timeout: float | Timeout | NotGiven = not_given) -> Any

List scans (host ip/fqdn, os, scan date, cvss score).

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of rows to return in this page.	30
offset	int	Number of rows to skip.	0
filter	dict[str, Any] | None	Additional report filter, if any.	None
sort	str	Sort field; prefix with - for descending.	''

Returns:

Type	Description
Any	The scan-list report payload: one row per scan.

Source code in src/vulners/_resources/_sync/report.py

def scan_list(
    self,
    *,
    limit: int = 30,
    offset: int = 0,
    filter: dict[str, Any] | None = None,
    sort: str = "",
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """List scans (host ip/fqdn, os, scan date, cvss score).

    Args:
        limit: Maximum number of rows to return in this page.
        offset: Number of rows to skip.
        filter: Additional report filter, if any.
        sort: Sort field; prefix with ``-`` for descending.

    Returns:
        The scan-list report payload: one row per scan.
    """
    return self._report("scanlist", limit, offset, filter, sort, timeout)

host_vulns

host_vulns(*, limit: int = 30, offset: int = 0, filter: dict[str, Any] | None = None, sort: str = '', timeout: float | Timeout | NotGiven = not_given) -> Any

List hosts with their cumulative fix and vulnerability ids.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of rows to return in this page.	30
offset	int	Number of rows to skip.	0
filter	dict[str, Any] | None	Additional report filter, if any.	None
sort	str	Sort field; prefix with - for descending.	''

Returns:

Type	Description
Any	The host-vulnerabilities report payload: one row per host with its
Any	fix and vulnerability ids.

Source code in src/vulners/_resources/_sync/report.py

def host_vulns(
    self,
    *,
    limit: int = 30,
    offset: int = 0,
    filter: dict[str, Any] | None = None,
    sort: str = "",
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """List hosts with their cumulative fix and vulnerability ids.

    Args:
        limit: Maximum number of rows to return in this page.
        offset: Number of rows to skip.
        filter: Additional report filter, if any.
        sort: Sort field; prefix with ``-`` for descending.

    Returns:
        The host-vulnerabilities report payload: one row per host with its
        fix and vulnerability ids.
    """
    return self._report("hostvulns", limit, offset, filter, sort, timeout)

vuln_info

vuln_info(ip_address: str, bulletin_id: str, *, limit: int = 30, offset: int = 0, filter: dict[str, Any] | None = None, sort: str = '', timeout: float | Timeout | NotGiven = not_given) -> Any

Detail of one vulnerability on one host.

Parameters:

Name	Type	Description	Default
ip_address	str	The host ip the vulnerability was found on.	required
bulletin_id	str	The vulnerability bulletin id (e.g. a CVE id).	required
limit	int	Maximum number of rows to return.	30
offset	int	Number of rows to skip.	0
filter	dict[str, Any] | None	Additional report filter, if any.	None
sort	str	Sort field; prefix with - for descending.	''

Source code in src/vulners/_resources/_sync/report.py

def vuln_info(
    self,
    ip_address: str,
    bulletin_id: str,
    *,
    limit: int = 30,
    offset: int = 0,
    filter: dict[str, Any] | None = None,
    sort: str = "",
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Detail of one vulnerability on one host.

    Args:
        ip_address: The host ip the vulnerability was found on.
        bulletin_id: The vulnerability bulletin id (e.g. a CVE id).
        limit: Maximum number of rows to return.
        offset: Number of rows to skip.
        filter: Additional report filter, if any.
        sort: Sort field; prefix with ``-`` for descending.
    """
    body: dict[str, Any] = {
        "reporttype": "vulninfo",
        "ipaddress": ip_address,
        "bulletinID": bulletin_id,
        "skip": offset,
        "size": limit,
        "filter": filter or {},
        "sort": sort,
    }
    return self._request(_REPORT, body=body, timeout=timeout)

Stix

vulners._resources._sync.stix.Stix

Stix(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

Build STIX bundles from Vulners bulletins.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

make_bundle_by_id

make_bundle_by_id(id: str, *, opencti_id: str | None = None, timeout: float | Timeout | NotGiven = not_given) -> Any

Build a STIX bundle of objects for a bulletin id.

Parameters:

Name	Type	Description	Default
id	str	The bulletin id to build a bundle for.	required
opencti_id	str | None	Existing OpenCTI object id to reuse, if any.	None

Returns:

Type	Description
Any	The decoded STIX bundle; when the server returns it as a JSON
Any	string the SDK re-parses it, so the result is always a decoded
Any	object.

Source code in src/vulners/_resources/_sync/stix.py

def make_bundle_by_id(
    self,
    id: str,
    *,
    opencti_id: str | None = None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Build a STIX bundle of objects for a bulletin id.

    Args:
        id: The bulletin id to build a bundle for.
        opencti_id: Existing OpenCTI object id to reuse, if any.

    Returns:
        The decoded STIX bundle; when the server returns it as a JSON
        string the SDK re-parses it, so the result is always a decoded
        object.
    """
    body: dict[str, Any] = {"id": id}
    if opencti_id is not None:
        body["opencti_id"] = opencti_id
    return self._request(_BUNDLE, cast=_parse_bundle, body=body, timeout=timeout)

bundle

bundle(id: str, *, opencti_id: str | None = None, timeout: float | Timeout | NotGiven = not_given) -> Any

Alias of :meth:make_bundle_by_id (the short primary name).

Source code in src/vulners/_resources/_sync/stix.py

def bundle(
    self,
    id: str,
    *,
    opencti_id: str | None = None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Alias of :meth:`make_bundle_by_id` (the short primary name)."""
    return self.make_bundle_by_id(id, opencti_id=opencti_id, timeout=timeout)

Subscriptions

vulners._resources._sync.subscriptions.Subscriptions

Subscriptions(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

Manage v3 email subscriptions.

The api_key argument on the mutating methods names the owner of the subscription (sent in the body as apiKey) and defaults to the client's own key. A privileged key sent in the X-Api-Key header can pass a different owner key to manage that key's subscriptions.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

list

list(*, timeout: float | Timeout | NotGiven = not_given) -> Any

List the email subscriptions registered under the client's API key.

Returns:

Type	Description
Any	The account's email subscriptions; each entry carries the
Any	subscription id used by :meth:edit and :meth:delete.

Source code in src/vulners/_resources/_sync/subscriptions.py

def list(
    self,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """List the email subscriptions registered under the client's API key.

    Returns:
        The account's email subscriptions; each entry carries the
        subscription id used by :meth:`edit` and :meth:`delete`.
    """
    return self._request(_LIST, timeout=timeout)

add

add(*, query: str, email: str, format: Literal['html', 'json', 'pdf'] = 'html', crontab: str | NotGiven = not_given, query_type: str = 'lucene', api_key: str | None = None, timeout: float | Timeout | NotGiven = not_given) -> Any

Create an email subscription for a query.

Parameters:

Name	Type	Description	Default
query	str	The search query to subscribe to.	required
email	str	Destination email address.	required
format	Literal['html', 'json', 'pdf']	Report format, "html", "json" or "pdf".	'html'
crontab	str | NotGiven	Optional crontab schedule.	not_given
query_type	str	Query language, defaults to "lucene".	'lucene'
api_key	str | None	Owner key for the subscription. Defaults to the client's own key; pass another key (with a privileged key on the client) to create the subscription under that key.	None

Source code in src/vulners/_resources/_sync/subscriptions.py

def add(
    self,
    *,
    query: str,
    email: str,
    format: Literal["html", "json", "pdf"] = "html",
    crontab: str | NotGiven = not_given,
    query_type: str = "lucene",
    api_key: str | None = None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Create an email subscription for a query.

    Args:
        query: The search query to subscribe to.
        email: Destination email address.
        format: Report format, ``"html"``, ``"json"`` or ``"pdf"``.
        crontab: Optional crontab schedule.
        query_type: Query language, defaults to ``"lucene"``.
        api_key: Owner key for the subscription. Defaults to the client's
            own key; pass another key (with a privileged key on the client)
            to create the subscription under that key.
    """
    body: dict[str, Any] = {
        "query": query,
        "email": email,
        "format": format,
        "query_type": query_type,
    }
    self._set(body, "crontab", crontab)
    body["apiKey"] = api_key or self._api_key
    return self._request(_ADD, body=body, timeout=timeout)

edit

edit(subscription_id: str, *, format: Literal['html', 'json', 'pdf'] | NotGiven = not_given, crontab: str | NotGiven = not_given, active: Literal['yes', 'no', 'true', 'false'] | NotGiven = not_given, api_key: str | None = None, timeout: float | Timeout | NotGiven = not_given) -> Any

Edit an existing email subscription.

Parameters:

Name	Type	Description	Default
subscription_id	str	The subscription to edit.	required
format	Literal['html', 'json', 'pdf'] | NotGiven	New report format, if changing.	not_given
crontab	str | NotGiven	New crontab schedule, if changing.	not_given
active	Literal['yes', 'no', 'true', 'false'] | NotGiven	New active state, if changing.	not_given
api_key	str | None	Owner key for the subscription (see :meth:add).	None

Source code in src/vulners/_resources/_sync/subscriptions.py

def edit(
    self,
    subscription_id: str,
    *,
    format: Literal["html", "json", "pdf"] | NotGiven = not_given,
    crontab: str | NotGiven = not_given,
    active: Literal["yes", "no", "true", "false"] | NotGiven = not_given,
    api_key: str | None = None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Edit an existing email subscription.

    Args:
        subscription_id: The subscription to edit.
        format: New report format, if changing.
        crontab: New crontab schedule, if changing.
        active: New active state, if changing.
        api_key: Owner key for the subscription (see :meth:`add`).
    """
    body: dict[str, Any] = {"subscriptionid": subscription_id}
    self._set(body, "format", format)
    self._set(body, "crontab", crontab)
    self._set(body, "active", active)
    body["apiKey"] = api_key or self._api_key
    return self._request(_EDIT, body=body, timeout=timeout)

delete

delete(subscription_id: str, *, api_key: str | None = None, timeout: float | Timeout | NotGiven = not_given) -> Any

Delete an email subscription.

Parameters:

Name	Type	Description	Default
subscription_id	str	The subscription to delete.	required
api_key	str | None	Owner key for the subscription (see :meth:add).	None

Source code in src/vulners/_resources/_sync/subscriptions.py

def delete(
    self,
    subscription_id: str,
    *,
    api_key: str | None = None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Delete an email subscription.

    Args:
        subscription_id: The subscription to delete.
        api_key: Owner key for the subscription (see :meth:`add`).
    """
    body = {"subscriptionid": subscription_id, "apiKey": api_key or self._api_key}
    return self._request(_DELETE, body=body, timeout=timeout)

SubscriptionsV4

vulners._resources._sync.subscriptions_v4.SubscriptionsV4

SubscriptionsV4(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

Manage v4 subscriptions.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

list

list(*, timeout: float | Timeout | NotGiven = not_given) -> Any

List every subscription on the account.

Returns:

Type	Description
Any	The account's subscriptions, one record per subscription.

Source code in src/vulners/_resources/_sync/subscriptions_v4.py

def list(
    self,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """List every subscription on the account.

    Returns:
        The account's subscriptions, one record per subscription.
    """
    return self._request(_LIST, timeout=timeout)

get_list

get_list(*, timeout: float | Timeout | NotGiven = not_given) -> Any

Alias of :meth:list, kept for the pre-release naming window.

Source code in src/vulners/_resources/_sync/subscriptions_v4.py

def get_list(
    self,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Alias of :meth:`list`, kept for the pre-release naming window."""
    return self.list(timeout=timeout)

get

get(id: str | None = None, *, subscription_id: str | None = None, timeout: float | Timeout | NotGiven = not_given) -> Any

Fetch a single subscription by id.

Parameters:

Name	Type	Description	Default
id	str | None	The subscription id. Interchangeable with subscription_id; pass exactly one of the two.	None
subscription_id	str | None	The subscription id under the server's query parameter name; interchangeable with id.	None

Raises:

Type	Description
TypeError	Neither or both of id and subscription_id were given.

Source code in src/vulners/_resources/_sync/subscriptions_v4.py

def get(
    self,
    id: str | None = None,
    *,
    subscription_id: str | None = None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Fetch a single subscription by id.

    Args:
        id: The subscription id. Interchangeable with ``subscription_id``;
            pass exactly one of the two.
        subscription_id: The subscription id under the server's query
            parameter name; interchangeable with ``id``.

    Raises:
        TypeError: Neither or both of ``id`` and ``subscription_id`` were
            given.
    """
    if (id is None) == (subscription_id is None):
        raise TypeError("get() requires exactly one of id= or subscription_id=")
    value = id if id is not None else subscription_id
    return self._request(_GET, body={"subscription_id": value}, timeout=timeout)

create

create(*, name: str, query: SubscriptionQuery | Mapping[str, Any], delivery: SubscriptionDelivery | Mapping[str, Any], license_id: str | None = None, bulletin_fields: Sequence[str] | None = None, is_active: bool = True, timestamp_source: TimestampSource = 'modified', send_empty_result: bool = False, timeout: float | Timeout | NotGiven = not_given) -> Any

Create a subscription.

Parameters:

Name	Type	Description	Default
name	str	Human-readable subscription name.	required
query	SubscriptionQuery | Mapping[str, Any]	Query definition, discriminated by type — a typed dict such as :class:~vulners._types.subscriptions.SubscriptionQueryLucene or a plain mapping.	required
delivery	SubscriptionDelivery | Mapping[str, Any]	Delivery definition, discriminated by type — a typed dict such as :class:~vulners._types.subscriptions.SubscriptionDeliveryWebhook or a plain mapping.	required
license_id	str | None	License to bill against, if any.	None
bulletin_fields	Sequence[str] | None	Bulletin fields to include in results.	None
is_active	bool	Whether the subscription starts active.	True
timestamp_source	TimestampSource	Which timestamp drives incremental delivery.	'modified'
send_empty_result	bool	Deliver even when there are no new results.	False

Source code in src/vulners/_resources/_sync/subscriptions_v4.py

def create(
    self,
    *,
    name: str,
    query: SubscriptionQuery | Mapping[str, Any],
    delivery: SubscriptionDelivery | Mapping[str, Any],
    license_id: str | None = None,
    bulletin_fields: Sequence[str] | None = None,
    is_active: bool = True,
    timestamp_source: TimestampSource = "modified",
    send_empty_result: bool = False,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Create a subscription.

    Args:
        name: Human-readable subscription name.
        query: Query definition, discriminated by ``type`` — a typed dict
            such as :class:`~vulners._types.subscriptions.SubscriptionQueryLucene`
            or a plain mapping.
        delivery: Delivery definition, discriminated by ``type`` — a typed
            dict such as
            :class:`~vulners._types.subscriptions.SubscriptionDeliveryWebhook`
            or a plain mapping.
        license_id: License to bill against, if any.
        bulletin_fields: Bulletin fields to include in results.
        is_active: Whether the subscription starts active.
        timestamp_source: Which timestamp drives incremental delivery.
        send_empty_result: Deliver even when there are no new results.
    """
    fields = (
        list(bulletin_fields)
        if bulletin_fields is not None
        else list(DEFAULT_BULLETIN_FIELDS)
    )
    body: dict[str, Any] = {
        "name": name,
        "query": dict(query),
        "delivery": dict(delivery),
        "licenseId": license_id,
        "bulletin_fields": fields,
        "is_active": is_active,
        "timestamp_source": timestamp_source,
        "send_empty_result": send_empty_result,
    }
    return self._request(_CREATE, body=body, timeout=timeout)

update

update(id: str, *, name: str | NotGiven = not_given, query: SubscriptionQuery | Mapping[str, Any] | NotGiven = not_given, delivery: SubscriptionDelivery | Mapping[str, Any] | NotGiven = not_given, license_id: str | NotGiven | None = not_given, bulletin_fields: Sequence[str] | NotGiven = not_given, is_active: bool | NotGiven = not_given, timestamp_source: TimestampSource | NotGiven = not_given, send_empty_result: bool | NotGiven = not_given, timeout: float | Timeout | NotGiven = not_given) -> Any

Update a subscription.

This is not a partial update. The client omits any argument you leave unset, but the server requires query, delivery and send_empty_result on every call and returns HTTP 400 if any of them is missing — so pass all three on every update, even when you only mean to change one other field.

Parameters:

Name	Type	Description	Default
id	str	The subscription to update.	required
name	str | NotGiven	New subscription name.	not_given
query	SubscriptionQuery | Mapping[str, Any] | NotGiven	New query definition (discriminated by type). Required on every call.	not_given
delivery	SubscriptionDelivery | Mapping[str, Any] | NotGiven	New delivery definition (discriminated by type). Required on every call.	not_given
license_id	str | NotGiven | None	License to bill against (None clears it).	not_given
bulletin_fields	Sequence[str] | NotGiven	Bulletin fields to include in results.	not_given
is_active	bool | NotGiven	Enable or disable the subscription.	not_given
timestamp_source	TimestampSource | NotGiven	Which timestamp drives incremental delivery.	not_given
send_empty_result	bool | NotGiven	Deliver even when there are no new results. Required on every call.	not_given

Source code in src/vulners/_resources/_sync/subscriptions_v4.py

def update(
    self,
    id: str,
    *,
    name: str | NotGiven = not_given,
    query: SubscriptionQuery | Mapping[str, Any] | NotGiven = not_given,
    delivery: SubscriptionDelivery | Mapping[str, Any] | NotGiven = not_given,
    license_id: str | NotGiven | None = not_given,
    bulletin_fields: Sequence[str] | NotGiven = not_given,
    is_active: bool | NotGiven = not_given,
    timestamp_source: TimestampSource | NotGiven = not_given,
    send_empty_result: bool | NotGiven = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Update a subscription.

    This is **not** a partial update. The client omits any argument you
    leave unset, but the server requires ``query``, ``delivery`` and
    ``send_empty_result`` on every call and returns HTTP 400 if any of them
    is missing — so pass all three on every update, even when you only mean
    to change one other field.

    Args:
        id: The subscription to update.
        name: New subscription name.
        query: New query definition (discriminated by ``type``). Required on
            every call.
        delivery: New delivery definition (discriminated by ``type``).
            Required on every call.
        license_id: License to bill against (``None`` clears it).
        bulletin_fields: Bulletin fields to include in results.
        is_active: Enable or disable the subscription.
        timestamp_source: Which timestamp drives incremental delivery.
        send_empty_result: Deliver even when there are no new results.
            Required on every call.
    """
    body: dict[str, Any] = {"id": id}
    self._set(body, "name", name)
    self._set(body, "query", query, dict)
    self._set(body, "delivery", delivery, dict)
    self._set(body, "licenseId", license_id)
    self._set(body, "bulletin_fields", bulletin_fields, list)
    self._set(body, "is_active", is_active)
    self._set(body, "timestamp_source", timestamp_source)
    self._set(body, "send_empty_result", send_empty_result)
    return self._request(_UPDATE, body=body, timeout=timeout)

delete

delete(id: str, *, timeout: float | Timeout | NotGiven = not_given) -> Any

Delete a subscription by id.

Parameters:

Name	Type	Description	Default
id	str	The subscription to delete.	required

Returns:

Type	Description
Any	The server's delete acknowledgement.

Source code in src/vulners/_resources/_sync/subscriptions_v4.py

def delete(
    self,
    id: str,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Delete a subscription by id.

    Args:
        id: The subscription to delete.

    Returns:
        The server's delete acknowledgement.
    """
    return self._request(_DELETE, body={"id": id}, timeout=timeout)

Webhooks

vulners._resources._sync.webhooks.Webhooks

Webhooks(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

Manage webhook subscriptions.

The api_key argument on the mutating methods is the owner of the subscription and defaults to the client's own key. To manage another key's subscriptions, authenticate the client with a privileged key and pass that other key as api_key.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

list

list(*, timeout: float | Timeout | NotGiven = not_given) -> Any

List the account's webhook subscriptions.

Scoped to the client's own key: the server reads this endpoint from the X-Api-Key header only, so there is no owner-key override here. Pair with :meth:add to create a subscription and :meth:read to poll one for pending payloads.

Source code in src/vulners/_resources/_sync/webhooks.py

def list(
    self,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """List the account's webhook subscriptions.

    Scoped to the client's own key: the server reads this endpoint from the
    ``X-Api-Key`` header only, so there is no owner-key override here. Pair
    with :meth:`add` to create a subscription and :meth:`read` to poll one
    for pending payloads.
    """
    return self._request(_LIST, timeout=timeout)

add

add(query: str, *, api_key: str | None = None, timeout: float | Timeout | NotGiven = not_given) -> Any

Create a webhook subscription for a query.

Parameters:

Name	Type	Description	Default
query	str	The Lucene query that defines matches.	required
api_key	str | None	Owner key for the subscription. Defaults to the client's own key; pass another key (with a privileged key on the client) to create the subscription under that key.	None

Source code in src/vulners/_resources/_sync/webhooks.py

def add(
    self,
    query: str,
    *,
    api_key: str | None = None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Create a webhook subscription for a query.

    Args:
        query: The Lucene query that defines matches.
        api_key: Owner key for the subscription. Defaults to the client's
            own key; pass another key (with a privileged key on the client)
            to create the subscription under that key.
    """
    body = {"query": query, "apiKey": api_key or self._api_key}
    return self._request(_ADD, body=body, timeout=timeout)

create

create(query: str, *, api_key: str | None = None, timeout: float | Timeout | NotGiven = not_given) -> Any

Alias of :meth:add (the primary CRUD-style name).

Source code in src/vulners/_resources/_sync/webhooks.py

def create(
    self,
    query: str,
    *,
    api_key: str | None = None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Alias of :meth:`add` (the primary CRUD-style name)."""
    return self.add(query, api_key=api_key, timeout=timeout)

enable

enable(id: str, active: bool, *, api_key: str | None = None, timeout: float | Timeout | NotGiven = not_given) -> Any

Enable or disable a webhook subscription.

Parameters:

Name	Type	Description	Default
id	str	The subscription to toggle.	required
active	bool	New active state.	required
api_key	str | None	Owner key for the subscription (see :meth:add).	None

Source code in src/vulners/_resources/_sync/webhooks.py

def enable(
    self,
    id: str,
    active: bool,
    *,
    api_key: str | None = None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Enable or disable a webhook subscription.

    Args:
        id: The subscription to toggle.
        active: New active state.
        api_key: Owner key for the subscription (see :meth:`add`).
    """
    body = {
        "subscriptionid": id,
        "active": "true" if active else "false",
        "apiKey": api_key or self._api_key,
    }
    return self._request(_EDIT, body=body, timeout=timeout)

set_enabled

set_enabled(id: str, active: bool, *, api_key: str | None = None, timeout: float | Timeout | NotGiven = not_given) -> Any

Alias of :meth:enable (the explicit setter-style name).

Source code in src/vulners/_resources/_sync/webhooks.py

def set_enabled(
    self,
    id: str,
    active: bool,
    *,
    api_key: str | None = None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Alias of :meth:`enable` (the explicit setter-style name)."""
    return self.enable(id, active, api_key=api_key, timeout=timeout)

read

read(id: str, *, newest_only: bool = True, api_key: str | None = None, timeout: float | Timeout | NotGiven = not_given) -> Any

Read pending webhook payloads for a subscription.

This endpoint requires the api key in the query string (the X-Api-Key header alone is rejected), so the key is echoed as a query parameter here.

Parameters:

Name	Type	Description	Default
id	str	The subscription to read.	required
newest_only	bool	Return only the newest stored payload.	True
api_key	str | None	Owner key for the subscription (see :meth:add).	None

Source code in src/vulners/_resources/_sync/webhooks.py

def read(
    self,
    id: str,
    *,
    newest_only: bool = True,
    api_key: str | None = None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Read pending webhook payloads for a subscription.

    This endpoint requires the api key in the query string (the ``X-Api-Key``
    header alone is rejected), so the key is echoed as a query parameter here.

    Args:
        id: The subscription to read.
        newest_only: Return only the newest stored payload.
        api_key: Owner key for the subscription (see :meth:`add`).
    """
    body = {
        "subscriptionid": id,
        "newest_only": "true" if newest_only else "false",
        "apiKey": api_key or self._api_key,
    }
    return self._request(_READ, body=body, timeout=timeout)

delete

delete(id: str, *, api_key: str | None = None, timeout: float | Timeout | NotGiven = not_given) -> Any

Delete a webhook subscription.

Parameters:

Name	Type	Description	Default
id	str	The subscription to delete.	required
api_key	str | None	Owner key for the subscription (see :meth:add).	None

Source code in src/vulners/_resources/_sync/webhooks.py

def delete(
    self,
    id: str,
    *,
    api_key: str | None = None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Delete a webhook subscription.

    Args:
        id: The subscription to delete.
        api_key: Owner key for the subscription (see :meth:`add`).
    """
    body = {"subscriptionid": id, "apiKey": api_key or self._api_key}
    return self._request(_DELETE, body=body, timeout=timeout)

VScanner

vulners._resources._sync.vscanner.Vscanner

Vscanner(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

VScanner product namespace on the client.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

licenses cached property

licenses: VscannerLicenses

Access to the account's VScanner license ids.

projects cached property

projects: VscannerProjects

VScanner projects, plus their nested tasks and results.

notification staticmethod

notification(period: Literal['disabled', 'asap', 'hourly', 'daily'], emails: Sequence[str] | None = None, slack_webhooks: Sequence[str] | None = None) -> dict[str, Any]

Build a notification object for a project.

Parameters:

Name	Type	Description	Default
period	Literal['disabled', 'asap', 'hourly', 'daily']	One of "disabled", "asap", "hourly", "daily".	required
emails	Sequence[str] | None	Email destinations.	None
slack_webhooks	Sequence[str] | None	Slack webhook destinations.	None

Returns:

Type	Description
dict[str, Any]	A notification object suitable for the notification argument of
dict[str, Any]	meth:VscannerProjects.create and
dict[str, Any]	meth:VscannerProjects.update.

Raises:

Type	Description
ValueError	If period is not one of the four accepted values.

Source code in src/vulners/_resources/_sync/vscanner.py

@staticmethod
def notification(
    period: Literal["disabled", "asap", "hourly", "daily"],
    emails: Sequence[str] | None = None,
    slack_webhooks: Sequence[str] | None = None,
) -> dict[str, Any]:
    """Build a notification object for a project.

    Args:
        period: One of ``"disabled"``, ``"asap"``, ``"hourly"``, ``"daily"``.
        emails: Email destinations.
        slack_webhooks: Slack webhook destinations.

    Returns:
        A notification object suitable for the ``notification`` argument of
        :meth:`VscannerProjects.create` and
        :meth:`VscannerProjects.update`.

    Raises:
        ValueError: If ``period`` is not one of the four accepted values.
    """
    if period not in ("disabled", "asap", "hourly", "daily"):
        raise ValueError(
            'period expected to be one of "disabled", "asap", "hourly" or "daily"'
        )
    return {
        "period": period,
        "email": list(emails or []),
        "slack": list(slack_webhooks or []),
    }

disabled_notification staticmethod

disabled_notification() -> dict[str, Any]

Build a notification object with delivery turned off.

Returns:

Type	Description
dict[str, Any]	A notification object with "disabled" period and no
dict[str, Any]	destinations, for a project that should send no alerts.

Source code in src/vulners/_resources/_sync/vscanner.py

@staticmethod
def disabled_notification() -> dict[str, Any]:
    """Build a notification object with delivery turned off.

    Returns:
        A notification object with ``"disabled"`` period and no
        destinations, for a project that should send no alerts.
    """
    return {"period": "disabled", "email": [], "slack": []}

The Vscanner namespace delegates to nested resources for project and license operations (v.vscanner.projects, v.vscanner.licenses); task and result operations are nested under a project (v.vscanner.projects.tasks, v.vscanner.projects.results):

vulners._resources._sync.vscanner.VscannerProjects

VscannerProjects(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

VScanner projects, plus their tasks and results namespaces.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

tasks cached property

tasks: VscannerTasks

Scan-task operations scoped to a project.

results cached property

results: VscannerResults

Scan-result and screenshot operations scoped to a project.

list

list(*, offset: int = 0, limit: int = 50, timeout: float | Timeout | NotGiven = not_given) -> Any

List the account's VScanner projects.

Parameters:

Name	Type	Description	Default
offset	int	Number of projects to skip.	0
limit	int	Maximum number of projects to return in this page.	50

Returns:

Type	Description
Any	A page of project records.

Source code in src/vulners/_resources/_sync/vscanner.py

def list(
    self,
    *,
    offset: int = 0,
    limit: int = 50,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """List the account's VScanner projects.

    Args:
        offset: Number of projects to skip.
        limit: Maximum number of projects to return in this page.

    Returns:
        A page of project records.
    """
    spec = _spec("GET", f"{_ROOT}/")
    return self._request(spec, body={"offset": offset, "limit": limit}, timeout=timeout)

create

create(*, name: str, license_id: UUID, notification: Mapping[str, Any], result_expire_in: int | NotGiven | None = not_given, timeout: float | Timeout | NotGiven = not_given) -> Any

Create a project.

Parameters:

Name	Type	Description	Default
name	str	New project name.	required
license_id	UUID	The license id to use.	required
notification	Mapping[str, Any]	A notification object (see :meth:Vscanner.notification).	required
result_expire_in	int | NotGiven | None	Expire results after N days; None never expires.	not_given

Returns:

Type	Description
Any	The created project record, including its assigned project id.

Source code in src/vulners/_resources/_sync/vscanner.py

def create(
    self,
    *,
    name: str,
    license_id: uuid.UUID,
    notification: Mapping[str, Any],
    result_expire_in: int | NotGiven | None = not_given,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Create a project.

    Args:
        name: New project name.
        license_id: The license id to use.
        notification: A notification object (see
            :meth:`Vscanner.notification`).
        result_expire_in: Expire results after N days; ``None`` never expires.

    Returns:
        The created project record, including its assigned project id.
    """
    body: dict[str, Any] = {
        "name": name,
        "license_id": str(license_id),
        "notification": dict(notification),
    }
    self._set(body, "result_expire_in", result_expire_in)
    spec = _spec("POST", f"{_ROOT}/", body_mode="json")
    return self._request(spec, body=body, timeout=timeout)

update

update(project_id: UUID, *, name: str, license_id: UUID, notification: Mapping[str, Any], result_expire_in: int | None, timeout: float | Timeout | NotGiven = not_given) -> Any

Replace a project's configuration in full.

Every field is required: unset fields overwrite the stored values rather than leaving them untouched.

Parameters:

Name	Type	Description	Default
project_id	UUID	The project to update.	required
name	str	Project name.	required
license_id	UUID	The license id to use.	required
notification	Mapping[str, Any]	A notification object (see :meth:Vscanner.notification).	required
result_expire_in	int | None	Expire results after N days; None never expires.	required

Returns:

Type	Description
Any	The updated project record.

Source code in src/vulners/_resources/_sync/vscanner.py

def update(
    self,
    project_id: uuid.UUID,
    *,
    name: str,
    license_id: uuid.UUID,
    notification: Mapping[str, Any],
    result_expire_in: int | None,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Replace a project's configuration in full.

    Every field is required: unset fields overwrite the stored values rather
    than leaving them untouched.

    Args:
        project_id: The project to update.
        name: Project name.
        license_id: The license id to use.
        notification: A notification object (see
            :meth:`Vscanner.notification`).
        result_expire_in: Expire results after N days; ``None`` never expires.

    Returns:
        The updated project record.
    """
    body = {
        "name": name,
        "license_id": str(license_id),
        "notification": dict(notification),
        "result_expire_in": result_expire_in,
    }
    spec = _spec("PUT", f"{_ROOT}/{_seg(project_id)}", body_mode="json")
    return self._request(spec, body=body, timeout=timeout)

delete

delete(project_id: UUID, *, timeout: float | Timeout | NotGiven = not_given) -> Any

Delete a project and its scan data.

Parameters:

Name	Type	Description	Default
project_id	UUID	The project to delete.	required

Returns:

Type	Description
Any	The API acknowledgement of the deletion.

Source code in src/vulners/_resources/_sync/vscanner.py

def delete(
    self,
    project_id: uuid.UUID,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Delete a project and its scan data.

    Args:
        project_id: The project to delete.

    Returns:
        The API acknowledgement of the deletion.
    """
    spec = _spec("DELETE", f"{_ROOT}/{_seg(project_id)}")
    return self._request(spec, timeout=timeout)

statistics

statistics(project_id: UUID, *, stat: Sequence[Literal['total_hosts', 'vulnerable_hosts', 'unique_cve', 'min_max_cvss', 'vulnerabilities_rank', 'vulnerable_hosts_rank']], timeout: float | Timeout | NotGiven = not_given) -> Any

Return project statistics for the requested aggregations.

Parameters:

Name	Type	Description	Default
project_id	UUID	The project to summarize.	required
stat	Sequence[Literal['total_hosts', 'vulnerable_hosts', 'unique_cve', 'min_max_cvss', 'vulnerabilities_rank', 'vulnerable_hosts_rank']]	Which aggregations to compute (e.g. "total_hosts", "unique_cve", "min_max_cvss").	required

Returns:

Type	Description
Any	A mapping keyed by the requested aggregation names, each holding its
Any	computed value.

Source code in src/vulners/_resources/_sync/vscanner.py

def statistics(
    self,
    project_id: uuid.UUID,
    *,
    stat: Sequence[
        Literal[
            "total_hosts",
            "vulnerable_hosts",
            "unique_cve",
            "min_max_cvss",
            "vulnerabilities_rank",
            "vulnerable_hosts_rank",
        ]
    ],
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Return project statistics for the requested aggregations.

    Args:
        project_id: The project to summarize.
        stat: Which aggregations to compute (e.g. ``"total_hosts"``,
            ``"unique_cve"``, ``"min_max_cvss"``).

    Returns:
        A mapping keyed by the requested aggregation names, each holding its
        computed value.
    """
    spec = _spec("GET", f"{_ROOT}/{_seg(project_id)}/statistic")
    return self._request(spec, body={"stat": list(stat)}, timeout=timeout)

vulners._resources._sync.vscanner.VscannerTasks

VscannerTasks(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

Scan tasks within a VScanner project.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

list

list(project_id: UUID, *, offset: int = 0, limit: int = 50, timeout: float | Timeout | NotGiven = not_given) -> Any

List the scan tasks defined in a project.

Parameters:

Name	Type	Description	Default
project_id	UUID	The owning project.	required
offset	int	Number of tasks to skip.	0
limit	int	Maximum number of tasks to return in this page.	50

Returns:

Type	Description
Any	A page of task records for the project.

Source code in src/vulners/_resources/_sync/vscanner.py

def list(
    self,
    project_id: uuid.UUID,
    *,
    offset: int = 0,
    limit: int = 50,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """List the scan tasks defined in a project.

    Args:
        project_id: The owning project.
        offset: Number of tasks to skip.
        limit: Maximum number of tasks to return in this page.

    Returns:
        A page of task records for the project.
    """
    spec = _spec("GET", f"{_ROOT}/{_seg(project_id)}/tasks")
    return self._request(spec, body={"offset": offset, "limit": limit}, timeout=timeout)

create

create(project_id: UUID, *, name: str, networks: Sequence[str], ports: Sequence[str], schedule: str, timing: str, enabled: bool, timeout: float | Timeout | NotGiven = not_given) -> Any

Create a scan task in a project.

Parameters:

Name	Type	Description	Default
project_id	UUID	The owning project.	required
name	str	Task name.	required
networks	Sequence[str]	Networks to scan (ips or domains).	required
ports	Sequence[str]	Ports or port ranges.	required
schedule	str	Crontab schedule string.	required
timing	str	Scan timing profile.	required
enabled	bool	Whether the task is enabled.	required

Returns:

Type	Description
Any	The created task record, including its assigned task id.

Source code in src/vulners/_resources/_sync/vscanner.py

def create(
    self,
    project_id: uuid.UUID,
    *,
    name: str,
    networks: Sequence[str],
    ports: Sequence[str],
    schedule: str,
    timing: str,
    enabled: bool,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Create a scan task in a project.

    Args:
        project_id: The owning project.
        name: Task name.
        networks: Networks to scan (ips or domains).
        ports: Ports or port ranges.
        schedule: Crontab schedule string.
        timing: Scan timing profile.
        enabled: Whether the task is enabled.

    Returns:
        The created task record, including its assigned task id.
    """
    body = {
        "name": name,
        "networks": list(networks),
        "ports": list(ports),
        "schedule": schedule,
        "timing": timing,
        "enabled": enabled,
    }
    spec = _spec("POST", f"{_ROOT}/{_seg(project_id)}/tasks", body_mode="json")
    return self._request(spec, body=body, timeout=timeout)

update

update(project_id: UUID, task_id: UUID, *, name: str, networks: Sequence[str], ports: Sequence[str], schedule: str, timing: str, enabled: bool, timeout: float | Timeout | NotGiven = not_given) -> Any

Replace a scan task's configuration in full.

Every field is required: unset fields overwrite the stored values rather than leaving them untouched.

Parameters:

Name	Type	Description	Default
project_id	UUID	The owning project.	required
task_id	UUID	The task to update.	required
name	str	Task name.	required
networks	Sequence[str]	Networks to scan (ips or domains).	required
ports	Sequence[str]	Ports or port ranges.	required
schedule	str	Crontab schedule string.	required
timing	str	Scan timing profile.	required
enabled	bool	Whether the task is enabled.	required

Returns:

Type	Description
Any	The updated task record.

Source code in src/vulners/_resources/_sync/vscanner.py

def update(
    self,
    project_id: uuid.UUID,
    task_id: uuid.UUID,
    *,
    name: str,
    networks: Sequence[str],
    ports: Sequence[str],
    schedule: str,
    timing: str,
    enabled: bool,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Replace a scan task's configuration in full.

    Every field is required: unset fields overwrite the stored values rather
    than leaving them untouched.

    Args:
        project_id: The owning project.
        task_id: The task to update.
        name: Task name.
        networks: Networks to scan (ips or domains).
        ports: Ports or port ranges.
        schedule: Crontab schedule string.
        timing: Scan timing profile.
        enabled: Whether the task is enabled.

    Returns:
        The updated task record.
    """
    body = {
        "name": name,
        "networks": list(networks),
        "ports": list(ports),
        "schedule": schedule,
        "timing": timing,
        "enabled": enabled,
    }
    spec = _spec("PUT", f"{_ROOT}/{_seg(project_id)}/tasks/{_seg(task_id)}", body_mode="json")
    return self._request(spec, body=body, timeout=timeout)

start

start(project_id: UUID, task_id: UUID, *, timeout: float | Timeout | NotGiven = not_given) -> Any

Queue a task to run as soon as possible, ignoring its schedule.

Parameters:

Name	Type	Description	Default
project_id	UUID	The owning project.	required
task_id	UUID	The task to start.	required

Returns:

Type	Description
Any	The API acknowledgement that the task was queued.

Source code in src/vulners/_resources/_sync/vscanner.py

def start(
    self,
    project_id: uuid.UUID,
    task_id: uuid.UUID,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Queue a task to run as soon as possible, ignoring its schedule.

    Args:
        project_id: The owning project.
        task_id: The task to start.

    Returns:
        The API acknowledgement that the task was queued.
    """
    spec = _spec(
        "POST", f"{_ROOT}/{_seg(project_id)}/tasks/{_seg(task_id)}/start", body_mode="json"
    )
    return self._request(spec, timeout=timeout)

delete

delete(project_id: UUID, task_id: UUID, *, timeout: float | Timeout | NotGiven = not_given) -> Any

Delete a scan task from a project.

Parameters:

Name	Type	Description	Default
project_id	UUID	The owning project.	required
task_id	UUID	The task to delete.	required

Returns:

Type	Description
Any	The API acknowledgement of the deletion.

Source code in src/vulners/_resources/_sync/vscanner.py

def delete(
    self,
    project_id: uuid.UUID,
    task_id: uuid.UUID,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Delete a scan task from a project.

    Args:
        project_id: The owning project.
        task_id: The task to delete.

    Returns:
        The API acknowledgement of the deletion.
    """
    spec = _spec("DELETE", f"{_ROOT}/{_seg(project_id)}/tasks/{_seg(task_id)}")
    return self._request(spec, timeout=timeout)

vulners._resources._sync.vscanner.VscannerResults

VscannerResults(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

Scan results and screenshots within a VScanner project.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

list

list(project_id: UUID, *, search: str | NotGiven = not_given, in_port: Sequence[str] | NotGiven = not_given, ex_port: Sequence[str] | NotGiven = not_given, min_cvss: float | NotGiven = not_given, max_cvss: float | NotGiven = not_given, last_seen: int | NotGiven = not_given, first_seen: int | NotGiven = not_given, last_seen_port: int | NotGiven = not_given, first_seen_port: int | NotGiven = not_given, sort: Literal['ip', 'name', 'last_seen', 'first_seen', 'resolved', 'min_cvss', 'max_cvss'] = 'last_seen', sort_dir: Literal['asc', 'desc'] = 'asc', offset: int = 0, limit: int = 50, timeout: float | Timeout | NotGiven = not_given) -> Any

List a project's scan results, with optional filtering and sorting.

Parameters:

Name	Type	Description	Default
project_id	UUID	The owning project.	required
search	str | NotGiven	Free-text query to match against results.	not_given
in_port	Sequence[str] | NotGiven	Keep only results on these ports.	not_given
ex_port	Sequence[str] | NotGiven	Drop results on these ports.	not_given
min_cvss	float | NotGiven	Keep only results with a CVSS score at or above this value.	not_given
max_cvss	float | NotGiven	Keep only results with a CVSS score at or below this value.	not_given
last_seen	int | NotGiven	Filter by the result's last-seen time (Unix timestamp).	not_given
first_seen	int | NotGiven	Filter by the result's first-seen time (Unix timestamp).	not_given
last_seen_port	int | NotGiven	Filter by a port's last-seen time (Unix timestamp).	not_given
first_seen_port	int | NotGiven	Filter by a port's first-seen time (Unix timestamp).	not_given
sort	Literal['ip', 'name', 'last_seen', 'first_seen', 'resolved', 'min_cvss', 'max_cvss']	Field to sort by.	'last_seen'
sort_dir	Literal['asc', 'desc']	Sort direction, ascending or descending.	'asc'
offset	int	Number of results to skip.	0
limit	int	Maximum number of results to return in this page.	50

Returns:

Type	Description
Any	A page of scan-result records matching the filters.

Source code in src/vulners/_resources/_sync/vscanner.py

def list(
    self,
    project_id: uuid.UUID,
    *,
    search: str | NotGiven = not_given,
    in_port: Sequence[str] | NotGiven = not_given,
    ex_port: Sequence[str] | NotGiven = not_given,
    min_cvss: float | NotGiven = not_given,
    max_cvss: float | NotGiven = not_given,
    last_seen: int | NotGiven = not_given,
    first_seen: int | NotGiven = not_given,
    last_seen_port: int | NotGiven = not_given,
    first_seen_port: int | NotGiven = not_given,
    sort: Literal[
        "ip", "name", "last_seen", "first_seen", "resolved", "min_cvss", "max_cvss"
    ] = "last_seen",
    sort_dir: Literal["asc", "desc"] = "asc",
    offset: int = 0,
    limit: int = 50,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """List a project's scan results, with optional filtering and sorting.

    Args:
        project_id: The owning project.
        search: Free-text query to match against results.
        in_port: Keep only results on these ports.
        ex_port: Drop results on these ports.
        min_cvss: Keep only results with a CVSS score at or above this value.
        max_cvss: Keep only results with a CVSS score at or below this value.
        last_seen: Filter by the result's last-seen time (Unix timestamp).
        first_seen: Filter by the result's first-seen time (Unix timestamp).
        last_seen_port: Filter by a port's last-seen time (Unix timestamp).
        first_seen_port: Filter by a port's first-seen time (Unix timestamp).
        sort: Field to sort by.
        sort_dir: Sort direction, ascending or descending.
        offset: Number of results to skip.
        limit: Maximum number of results to return in this page.

    Returns:
        A page of scan-result records matching the filters.
    """
    body: dict[str, Any] = {
        "sort": sort,
        "sort_dir": sort_dir,
        "offset": offset,
        "limit": limit,
    }
    self._set(body, "search", search)
    self._set(body, "in_port", in_port, list)
    self._set(body, "ex_port", ex_port, list)
    self._set(body, "min_cvss", min_cvss)
    self._set(body, "max_cvss", max_cvss)
    self._set(body, "last_seen", last_seen)
    self._set(body, "first_seen", first_seen)
    self._set(body, "last_seen_port", last_seen_port)
    self._set(body, "first_seen_port", first_seen_port)
    spec = _spec("GET", f"{_ROOT}/{_seg(project_id)}/results")
    return self._request(spec, body=body, timeout=timeout)

delete

delete(project_id: UUID, result_id: UUID, *, timeout: float | Timeout | NotGiven = not_given) -> Any

Delete a single scan result from a project.

Parameters:

Name	Type	Description	Default
project_id	UUID	The owning project.	required
result_id	UUID	The scan result to delete.	required

Returns:

Type	Description
Any	The API acknowledgement of the deletion.

Source code in src/vulners/_resources/_sync/vscanner.py

def delete(
    self,
    project_id: uuid.UUID,
    result_id: uuid.UUID,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Delete a single scan result from a project.

    Args:
        project_id: The owning project.
        result_id: The scan result to delete.

    Returns:
        The API acknowledgement of the deletion.
    """
    spec = _spec("DELETE", f"{_ROOT}/{_seg(project_id)}/results/{_seg(result_id)}")
    return self._request(spec, timeout=timeout)

screenshot

screenshot(image_uri: str, *, as_base64: bool = False, timeout: float | Timeout | NotGiven = not_given) -> bytes

Download a result screenshot as bytes.

Parameters:

Name	Type	Description	Default
image_uri	str	The server-provided screenshot uri (from a result's screens). It is validated to stay under /vscanner/screen/.	required
as_base64	bool	Return base64-encoded bytes instead of raw bytes.	False

Returns:

Type	Description
bytes	The screenshot image bytes, base64-encoded when as_base64 is set.

Raises:

Type	Description
ValueError	If image_uri resolves outside /vscanner/screen/ or uses excessive percent-encoding.

Source code in src/vulners/_resources/_sync/vscanner.py

def screenshot(
    self,
    image_uri: str,
    *,
    as_base64: bool = False,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> bytes:
    """Download a result screenshot as bytes.

    Args:
        image_uri: The server-provided screenshot uri (from a result's
            ``screens``). It is validated to stay under ``/vscanner/screen/``.
        as_base64: Return base64-encoded bytes instead of raw bytes.

    Returns:
        The screenshot image bytes, base64-encoded when ``as_base64`` is set.

    Raises:
        ValueError: If ``image_uri`` resolves outside ``/vscanner/screen/``
            or uses excessive percent-encoding.
    """
    url = "/vscanner/screen/" + image_uri
    _guard_screenshot_path(url)
    spec = _spec("GET", url, response_mode="bytes")
    content: bytes = self._request(spec, timeout=timeout)
    return base64.b64encode(content) if as_base64 else content

vulners._resources._sync.vscanner.VscannerLicenses

VscannerLicenses(client: SyncAPIClient, _wrap: str | None = None)

Bases: BaseResource

VScanner license ids.

Source code in src/vulners/_resources/_sync/_base.py

def __init__(self, client: SyncAPIClient, _wrap: str | None = None) -> None:
    self._client = client
    self._wrap = _wrap

list

list(*, timeout: float | Timeout | NotGiven = not_given) -> Any

List the account's VScanner license ids.

Returns:

Type	Description
Any	The license ids available to the account; pass one as license_id
Any	when creating a project (see :meth:VscannerProjects.create).

Source code in src/vulners/_resources/_sync/vscanner.py

def list(
    self,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """List the account's VScanner license ids.

    Returns:
        The license ids available to the account; pass one as ``license_id``
        when creating a project (see :meth:`VscannerProjects.create`).
    """
    return self._request(_LICENSES, timeout=timeout)