attestd

Attestd checks whether a dependency version has exploitable CVEs or a confirmed supply-chain compromise. One API call returns a structured risk response.

Get a free API key · Full docs

Requires Python 3.10+.

Install

pip install attestd

Quick start

import os
import attestd

client = attestd.Client(api_key=os.environ["ATTESTD_API_KEY"])
result = client.check("nginx", "1.20.0")
print(result.risk_state)  # "high"
print(result.cve_ids)     # ["CVE-2021-23017", ...]

Batch check

Check up to 100 packages in one request. Each item costs one API call. A 429 rejects the entire batch before billing.

results = client.batch_check([
    ("litellm", "1.82.7"),
    ("nginx", "1.25.3"),
])
# results[i] is RiskResult | None — None means outside coverage

Async:

results = await client.batch_check([("litellm", "1.82.7"), ("nginx", "1.25.3")])

Per-CVE detail

Default check / batch_check omit include and return compact results (cves is []). Pass include=["cves"] to request CVSS and EPSS per CVE. Compact and detailed responses are cached separately.

detailed = client.check("nginx", "1.20.0", include=["cves"])
print(detailed.cves[0].cvss_score)
print(detailed.cves[0].epss_score)

batch = client.batch_check(
    [("nginx", "1.20.0")],
    include=["cves"],
)

Catalog and quota

Three additional endpoints for product discovery, CVE lookup, and quota monitoring. All require a valid API key.

Products catalog

catalog = client.products()
print(catalog.total)
print(catalog.cve_products[0].slug)
print(catalog.supply_chain_packages[0].package)

Returns ProductsResult with cve_products, supply_chain_packages, and total. Maps to GET /v1/products.

CVE detail

try:
    detail = client.cve("CVE-2021-44228")
    print(detail.cvss_score, detail.epss_score)
except attestd.AttestdAPIError as e:
    if e.status_code == 404:
        print("CVE not in database")

Returns CveDetail with CVSS, EPSS, KEV status, and affected products. Raises AttestdAPIError(status_code=404) when the CVE is not in Attestd's database.

Usage quota

usage = client.usage()
print(usage.tier)
print(usage.key_calls_this_month, "/", usage.included_calls)
print(usage.billing_period_end)

Returns UsageResult with calls used this month, included cap, billing period, and overage estimate. Use for quota monitoring in CI or agent loops.

Supply chain check

Attestd monitors select PyPI and npm packages for known malicious publishes.

result = client.check("litellm", "1.82.7")
if result.supply_chain:
    print(result.supply_chain.compromised)  # True

Error handling

AttestdUnsupportedProductError means the product is outside Attestd coverage. That is unknown risk, not a safety signal.

try:
    result = client.check(product, version)
except attestd.AttestdUnsupportedProductError:
    raise RuntimeError(f"{product} is outside Attestd coverage")

See Outside coverage below for policy options.

Async

import asyncio
import os
import attestd

async def main():
    async with attestd.AsyncClient(api_key=os.environ["ATTESTD_API_KEY"]) as client:
        result = await client.check("log4j", "2.14.1")
        if result.risk_state in ("critical", "high"):
            raise RuntimeError(f"Vulnerable dependency: {result.cve_ids}")

asyncio.run(main())

CI/CD deployment gate

import os
import attestd

DEPENDENCIES = [
    ("nginx", "1.20.0"),
    ("log4j", "2.17.1"),
    ("openssh", "9.2p1"),
]

with attestd.Client(api_key=os.environ["ATTESTD_API_KEY"]) as client:
    for product, version in DEPENDENCIES:
        try:
            result = client.check(product, version)
        except attestd.AttestdUnsupportedProductError:
            continue

        if result.risk_state in ("critical", "high"):
            print(f"BLOCK: {product} {version} ({result.risk_state})")
            print(f"  CVEs: {', '.join(result.cve_ids)}")
            print(f"  Fix:  upgrade to {result.fixed_version}")
            exit(1)

AI agent tool

import os
import attestd

client = attestd.Client(api_key=os.environ["ATTESTD_API_KEY"])

def check_dependency_risk(product: str, version: str) -> dict:
    """Return risk_state, CVE IDs, and fixed version for a dependency."""
    try:
        result = client.check(product, version)
        return {
            "supported": True,
            "risk_state": result.risk_state,
            "actively_exploited": result.actively_exploited,
            "fixed_version": result.fixed_version,
            "cve_ids": result.cve_ids,
        }
    except attestd.AttestdUnsupportedProductError:
        return {"supported": False}

Full error handling

import os
import time
import attestd

with attestd.Client(api_key=os.environ["ATTESTD_API_KEY"]) as client:
    try:
        result = client.check("nginx", "1.20.0")
    except attestd.AttestdUnsupportedProductError:
        pass
    except attestd.AttestdRateLimitError as e:
        time.sleep(e.retry_after or 60)
    except attestd.AttestdAuthError:
        raise
    except attestd.AttestdAPIError as e:
        print(f"API error: {e.status_code}")
    except attestd.AttestdError:
        pass

Outside coverage: not a safety signal

AttestdUnsupportedProductError means Attestd has no vulnerability data for this product, not that the product is free of vulnerabilities. An agent that catches the exception and treats it as safe to proceed is making a dangerous inference.

Recommended handling:

try:
    result = client.check(product, version)
except attestd.AttestdUnsupportedProductError as e:
    raise RuntimeError(
        f"{e.product} is outside Attestd's coverage. "
        "Manual security review required before deploying."
    )

RiskResult fields

TyposquatSignal: detected, resembles, confidence, ecosystem, kind, likely_intended

CveSummary: cve_id, cvss_score, actively_exploited, remote_exploitable, epss_score, epss_percentile

Risk states

Risk factors

Catalog and quota types

Configuration

client = attestd.Client(
    api_key=os.environ.get("ATTESTD_API_KEY"),
    base_url="https://api.attestd.io",
    timeout=10.0,
    max_retries=3,
    retry_delay=1.0,
    cache_policy="runtime",
)
print(client.stats())
client.invalidate_cache("nginx", "1.20.0")

cache_policy is one of development, runtime, ci, or none. Default is runtime. stats() returns session counters (api_calls_made, cache_hits, batch_saves, calls_saved). invalidate_cache(product, version) drops both the compact and detailed (include=["cves"]) cached results for that product and version.

AsyncClient accepts the same options plus batch_window_ms (default 5). Concurrent check() calls within that window are coalesced into one batch request. Set batch_window_ms=0 to disable coalescing.

Set ATTESTD_API_KEY in your environment from the portal. Then omit api_key in the constructor:

with attestd.Client() as client:
    result = client.check("nginx", "1.20.0")

The SDK retries on transient 5xx responses and connection failures with exponential backoff (1s, 2s, 4s between attempts). Timeouts raise immediately and are not retried. 401 and 429 are surfaced immediately without retry.

Supported products

See attestd.io/docs/products for the full list. Each product page documents the API slug, version format, and notable CVEs. Querying an unsupported product raises AttestdUnsupportedProductError.

Testing your integration

The SDK ships an attestd.testing module with httpx transports for injecting controlled API responses. No local Attestd instance required.

import attestd
from attestd.testing import (
    MockTransport,
    MockAsyncTransport,
    SequentialMockTransport,
    SequentialMockAsyncTransport,
    NGINX_SAFE,
    NGINX_VULNERABLE,
    LOG4J_CRITICAL,
    UNSUPPORTED,
    LITELLM_SAFE,
    LITELLM_COMPROMISED,
    PRODUCTS_RESPONSE,
    CVE_LOG4SHELL,
    USAGE_SOLO,
    STATS_EMPTY,
)

Test a deployment gate

from attestd.testing import MockTransport, LOG4J_CRITICAL

def test_deployment_blocked_on_critical():
    client = attestd.Client(
        api_key="test",
        transport=MockTransport(200, LOG4J_CRITICAL),
    )
    with pytest.raises(DeploymentBlockedError):
        run_deployment_gate(client, "log4j", "2.14.1")

Test retry behaviour

from attestd.testing import SequentialMockTransport, NGINX_SAFE

def test_retry_succeeds_on_second_attempt():
    transport = SequentialMockTransport([
        (503, {}),
        (200, NGINX_SAFE),
    ])
    client = attestd.Client(api_key="test", transport=transport, max_retries=1)
    result = client.check("nginx", "1.27.4")
    assert result.risk_state == "none"
    assert transport.call_count == 2

Test the outside coverage policy branch

from attestd.testing import MockTransport, UNSUPPORTED

def test_outside_coverage_is_blocked_not_allowed():
    client = attestd.Client(api_key="test", transport=MockTransport(200, UNSUPPORTED))
    with pytest.raises(attestd.AttestdUnsupportedProductError):
        run_check(client, "unknownproduct", "1.0.0")

Custom response bodies

All ready-made bodies (NGINX_SAFE, etc.) are plain dicts. Merge in overrides:

from attestd.testing import MockTransport, NGINX_VULNERABLE

body = {**NGINX_VULNERABLE, "actively_exploited": True, "risk_state": "critical"}
client = attestd.Client(api_key="test", transport=MockTransport(200, body))

License

MIT