Platform Embed Integration

Your platform request handler
         │
         ▼
attestd.check(product, version)   ← before the irreversible action
         │
         ├─ risk_state: critical | high          → block
         ├─ supply_chain.compromised: true       → block
         ├─ typosquat.detected: true             → reject + surface resembles
         ├─ outside_coverage / unsupported       → apply your fallback policy
         └─ risk_state: none (and clean SC)      → allow

curl -X POST "https://api.attestd.io/v1/check/batch" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "items": [
      { "product": "log4j", "version": "2.14.1" },
      { "product": "nginx", "version": "1.20.0" },
      { "product": "litellm", "version": "1.82.7" }
    ]
  }'

import os
import attestd
from attestd import AttestdAPIError, AttestdUnsupportedProductError

# fail_closed=True blocks when Attestd is unreachable or returns 5xx.
# fail_closed=False (fail-open) allows the action and logs a warning.
FAIL_CLOSED = os.environ.get("ATTESTD_FAIL_CLOSED", "true").lower() == "true"

client = attestd.Client(
    api_key=os.environ["ATTESTD_API_KEY"],
    cache_policy="runtime",  # 5-minute TTL; cache hits do not bill
)

def gate(product: str, version: str) -> None:
    try:
        result = client.check(product, version)
    except AttestdUnsupportedProductError as e:
        if e.typosquat and e.typosquat.detected:
            raise PermissionError(
                f"name integrity failed for {product}: "
                f"prefer {e.typosquat.resembles}"
            )
        # Outside coverage: treat as unknown risk.
        if FAIL_CLOSED:
            raise PermissionError(f"outside coverage: {product}")
        return
    except AttestdAPIError as e:
        # 429, 5xx, timeout, network. Decide per your policy.
        if FAIL_CLOSED:
            raise PermissionError(f"attestd unavailable: {e}") from e
        return

    if result.risk_state in ("critical", "high"):
        raise PermissionError(f"{product}@{version} risk_state={result.risk_state}")
    if result.supply_chain and result.supply_chain.compromised:
        raise PermissionError(f"{product}@{version} supply chain compromised")

"""Agent security gateway: call before an agent installs or executes a package."""
import os
import attestd
from attestd import AttestdUnsupportedProductError

class SecurityGatewayError(Exception):
    pass

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

def allow_install(product: str, version: str) -> None:
    """Raise SecurityGatewayError if the package must not be installed."""
    try:
        result = client.check(product, version)
    except AttestdUnsupportedProductError as e:
        if e.typosquat and e.typosquat.detected:
            raise SecurityGatewayError(
                f"hallucinated or typosquat name '{product}'. "
                f"prefer {e.typosquat.resembles} "
                f"(likely_intended={list(e.typosquat.likely_intended)})"
            )
        raise SecurityGatewayError(
            f"no Attestd coverage for '{product}'. treat as unknown risk"
        )

    if result.supply_chain and result.supply_chain.compromised:
        raise SecurityGatewayError(
            f"{product}@{version} supply_chain.compromised=true "
            f"sources={result.supply_chain.sources}"
        )

    if result.risk_state in ("critical", "high"):
        raise SecurityGatewayError(
            f"{product}@{version} risk_state={result.risk_state} "
            f"fixed_version={result.fixed_version}"
        )

# Example: agent about to install a dependency
allow_install("log4j", "2.14.1")  # raises: risk_state=critical

/** Sandbox provider: pre-flight before admitting a dependency into the sandbox. */
import { Client, AttestdAPIError, AttestdUnsupportedProductError } from '@attestd/sdk';

const client = new Client({
  apiKey: process.env.ATTESTD_API_KEY!,
  cachePolicy: 'runtime',
});

export type PreflightResult = {
  allowed: boolean;
  reason: string;
  warnings: string[];
};

export async function preflightDependency(
  product: string,
  version: string,
): Promise<PreflightResult> {
  const warnings: string[] = [];

  try {
    const result = await client.check(product, version);

    if (result.supplyChain?.compromised) {
      return {
        allowed: false,
        reason: `supply_chain.compromised for ${product}@${version}`,
        warnings,
      };
    }

    if (result.riskState === 'critical' || result.riskState === 'high') {
      return {
        allowed: false,
        reason: `risk_state=${result.riskState} for ${product}@${version}`,
        warnings,
      };
    }

    // npm provenance: false means the package has a baseline but this version
    // lacks attestation. Surface as a warning; do not hard-block by default.
    if (result.supplyChain?.provenance === false) {
      warnings.push(
        `provenance=false for ${product}@${version}: version lacks attestation`,
      );
    }

    return { allowed: true, reason: 'ok', warnings };
  } catch (err) {
    if (err instanceof AttestdUnsupportedProductError) {
      if (err.typosquat?.detected) {
        return {
          allowed: false,
          reason: `name integrity failed; prefer ${err.typosquat.resembles}`,
          warnings,
        };
      }
      return {
        allowed: false,
        reason: `outside coverage for ${product}`,
        warnings,
      };
    }
    if (err instanceof AttestdAPIError) {
      // Propagate so the caller can apply fail-open / fail-closed.
      throw err;
    }
    throw err;
  }
}

/** Admit many packages at once (max 100 per batch call). */
export async function preflightBatch(
  deps: { product: string; version: string }[],
) {
  const results = await client.checkBatch(deps);
  return deps.map((d, i) => ({ dep: d, result: results[i] }));
}

"""Automation platform CI gate: fail the pipeline on critical risk or compromise."""
import os
import sys
import attestd

# Comma-separated risk_state values that fail the build.
# Default: critical,high. Set ATTESTD_FAIL_ON=critical to only block critical.
FAIL_ON = set(
    os.environ.get("ATTESTD_FAIL_ON", "critical,high").lower().split(",")
)

# Parsed from your SBOM / lockfile. Each item is (product, version).
DEPS = [
    ("nginx", "1.20.0"),
    ("log4j", "2.25.5"),
    ("litellm", "1.82.7"),
]

client = attestd.Client(
    api_key=os.environ["ATTESTD_API_KEY"],
    cache_policy="ci",  # infinite TTL within a single CI run
)

failed = False
results = client.batch_check(DEPS)

for (product, version), result in zip(DEPS, results):
    if result is None:
        print(f"  FAIL [OUTSIDE] {product}@{version}", file=sys.stderr)
        failed = True
        continue

    sc = result.supply_chain
    if sc is not None and sc.compromised:
        print(
            f"  FAIL [COMPROMISED] {product}@{version} sources={sc.sources}",
            file=sys.stderr,
        )
        failed = True
        continue

    if result.risk_state in FAIL_ON:
        print(
            f"  FAIL [{result.risk_state.upper()}] {product}@{version} "
            f"fixed_version={result.fixed_version}",
            file=sys.stderr,
        )
        failed = True
        continue

    print(f"  PASS [{result.risk_state}] {product}@{version}")

sys.exit(1 if failed else 0)