Skip to content

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.

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. When fields is omitted, the compact :data:DEFAULT_SEARCH_FIELDS projection is used (heavy fields like sourceData/description are opt-in via fields=["*"] or an explicit list).

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. When ``fields`` is
    omitted, the compact :data:`DEFAULT_SEARCH_FIELDS` projection is used
    (heavy fields like ``sourceData``/``description`` are opt-in via
    ``fields=["*"]`` or an explicit list).
    """
    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.

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`.
    """
    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.

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`.
    """
    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', timeout: float | Timeout | NotGiven = not_given) -> list[dict[str, Any]]

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

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'
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",
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> list[dict[str, Any]]:
    """Audit a list of software (CPE dicts or strings) for vulnerabilities.

    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.
    """
    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)
    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', timeout: float | Timeout | NotGiven = not_given) -> list[dict[str, Any]]

Audit a host: its software plus optional application/OS/hardware CPEs.

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",
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> list[dict[str, Any]]:
    """Audit a host: its software plus optional application/OS/hardware CPEs."""
    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)
    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, timeout: float | Timeout | NotGiven = not_given) -> dict[str, Any]

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

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 (non-free licenses only).

False
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,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, Any]:
    """Audit RPM/DEB/APK package lists for a Linux host.

    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 (non-free licenses only).
    """
    body = {
        "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,
    }
    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, timeout: float | Timeout | NotGiven = not_given) -> dict[str, Any]

Audit a list of packages in PURL format.

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 (non-free licenses only).

False
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,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, Any]:
    """Audit a list of packages in PURL format.

    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 (non-free licenses only).
    """
    body = {
        "packages": _require_count(
            packages, "packages", hi=_AUDIT_MAX_PACKAGES, strings=True
        ),
        "includeUnofficial": include_unofficial,
        "includeCandidates": include_candidates,
        "includeAnyVersion": include_any_version,
        "cvelistMetrics": cvelist_metrics,
    }
    return self._request(_LIBRARY, body=body, timeout=timeout)

sbom_audit

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

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

Parameters:

Name Type Description Default
file str | PathLike[str]

Path to the SBOM file to upload.

required
Source code in src/vulners/_resources/_sync/audit.py
def sbom_audit(
    self,
    file: str | os.PathLike[str],
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, Any]:
    """Audit an SBOM file (SPDX or CycloneDX) for vulnerabilities.

    Args:
        file: Path to the SBOM file to upload.
    """
    path = os.fspath(file)
    content = vulners._base_client._call_blocking(_read_file_bytes, path)
    files = {"file": (os.path.basename(path), content, "application/json")}
    return self._request(_SBOM, files=files, timeout=timeout)

cve_audit

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

Audit a single CVE identifier.

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."""
    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 (at least one).

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 (at least one)."""
    return self._request(
        _CVES, body={"cve": _require_count(cve, "cve", strings=True)}, timeout=timeout
    )

kb_audit

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

Audit a Windows host by its installed KB list.

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(
    self,
    os: str,
    kb_list: Sequence[str],
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> dict[str, Any]:
    """Audit a Windows host by its installed KB list.

    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, 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', 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", "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'

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",
    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", "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.

    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}
    return self._request(_SMART, body=body, 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").

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"``)."""
    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. Each element is a dict; records delivered as raw Elasticsearch hits are normalized to their "_source" document.

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. Each element is a ``dict``;
    records delivered as raw Elasticsearch hits are normalized to their
    ``"_source"`` document.
    """
    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, timeout: float | Timeout | NotGiven = not_given) -> int

Stream a collection archive straight to path; return bytes written.

The raw (still-compressed) archive bytes are written to path chunk by chunk — nothing is buffered in memory and nothing is decompressed — so a multi-gigabyte collection downloads in constant memory. Pass update_from to fetch only the entries changed after that moment (the collection-update endpoint) instead of the whole collection.

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.

required
update_from datetime | None

When given, download the collection update since this moment instead of the full archive.

None

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,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> int:
    """Stream a collection archive straight to ``path``; return bytes written.

    The raw (still-compressed) archive bytes are written to ``path`` chunk by
    chunk — nothing is buffered in memory and nothing is decompressed — so a
    multi-gigabyte collection downloads in constant memory. Pass
    ``update_from`` to fetch only the entries changed after that moment (the
    collection-update endpoint) instead of the whole collection.

    Args:
        collection: The collection type to download (e.g. ``"cve"``).
        path: Destination file path; an existing file is overwritten.
        update_from: When given, download the collection update since this
            moment instead of the full archive.

    Returns:
        The number of bytes written to ``path``.
    """
    spec = _STREAM_COLLECTION
    params: dict[str, Any] = {"type": collection}
    if update_from is not None:
        spec = _STREAM_COLLECTION_UPDATE
        params["after"] = update_from.isoformat()
    dest = os.fspath(path)
    # Atomic write: stream into a temp file beside the destination, fsync, then
    # os.replace() (an atomic same-filesystem rename). A timeout, cancellation,
    # disconnect or ENOSPC mid-download leaves any existing archive at `path`
    # intact instead of clobbering it with a half-written file.
    handle, tmp_path = vulners._base_client._call_blocking(_open_temp_beside, dest)
    written = 0
    try:
        with self._client.stream_response(spec, params=params, timeout=timeout) as resp:
            for chunk in resp.iter_bytes():
                vulners._base_client._call_blocking(handle.write, chunk)
                written += len(chunk)
        vulners._base_client._call_blocking(_flush_and_close, handle)
        vulners._base_client._call_blocking(os.replace, tmp_path, dest)
    except BaseException:
        vulners._base_client._call_blocking(_cleanup_temp, handle, tmp_path)
        raise
    return written

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.

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``."""
    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.

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.
    """
    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).

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)."""
    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. Pass update_from to stream only the entries changed after that moment (must be at most 25 hours ago) instead of the whole family.

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. Pass ``update_from`` to stream
    only the entries changed after that moment (must be at most 25 hours
    ago) instead of the whole family.
    """
    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).

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)."""
    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).

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)."""
    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).

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)."""
    return self._request(_GETSPLOIT, 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
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).
    """
    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.

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.
    """
    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'
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.
    """
    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.

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."""
    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

Summary of every found vulnerability (id, title, score, severity...).

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:
    """Summary of every found vulnerability (id, title, score, severity...)."""
    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 of vulnerabilities found on hosts, with host information.

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 of vulnerabilities found on hosts, with host information."""
    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

Per-host summary (agent id, ip, fqdn, os, vulnerability counts).

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:
    """Per-host summary (agent id, ip, fqdn, os, vulnerability counts)."""
    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 of scans (host ip/fqdn, os, scan date, cvss score).

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 of scans (host ip/fqdn, os, scan date, cvss score)."""
    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 of hosts with their cumulative 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 of hosts with their cumulative 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
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.
    """
    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 account's email subscriptions.

Source code in src/vulners/_resources/_sync/subscriptions.py
def list(
    self,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """List the account's email subscriptions."""
    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.

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."""
    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 | None | NotGiven = 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 | None | NotGiven

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 | None | NotGiven = 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.

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."""
    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.

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.
    """
    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

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
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.
    """
    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 stub notification object with "disabled" period.

Source code in src/vulners/_resources/_sync/vscanner.py
@staticmethod
def disabled_notification() -> dict[str, Any]:
    """Build a stub notification object with ``"disabled"`` period."""
    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

list

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

List the account's projects.

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 projects."""
    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 | None | NotGiven = 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 | None | NotGiven

Expire results after N days; None never expires.

not_given
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 | None | NotGiven = 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.
    """
    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

Update a project (full replace).

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:
    """Update a project (full replace)."""
    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.

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."""
    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.

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."""
    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 a project's tasks.

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 a project's tasks."""
    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
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.
    """
    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

Update a scan task (full replace).

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:
    """Update a scan task (full replace)."""
    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

Start a task as soon as possible.

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:
    """Start a task as soon as possible."""
    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 task.

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 task."""
    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.

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."""
    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.

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."""
    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
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.
    """
    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

Return the account's VScanner license ids.

Source code in src/vulners/_resources/_sync/vscanner.py
def list(
    self,
    *,
    timeout: float | httpx.Timeout | NotGiven = not_given,
) -> Any:
    """Return the account's VScanner license ids."""
    return self._request(_LICENSES, timeout=timeout)