KYA SDK Quickstart

# Coming soon: not published yet
npm install @baselayer/kya

# Coming soon: not published yet
pip install baselayer-kya

import { InMemoryNonceStore, KyaVerifier } from "@baselayer/kya";

const verifier = new KyaVerifier({
  // Your domain, exactly as credentials are minted for you.
  audience: "shop.example",
  // One process only. Several servers need a shared store.
  nonceStore: new InMemoryNonceStore(),
  // Sandbox credentials are test data and are refused by default.
  // allowSandbox: true,
});

from baselayer_kya import InMemoryNonceStore, KyaVerifier

verifier = KyaVerifier(
    # Your domain, exactly as credentials are minted for you.
    audience="shop.example",
    # One process only. Several servers need a shared store.
    nonce_store=InMemoryNonceStore(),
    # Sandbox credentials are test data and are refused by default.
    # allow_sandbox=True,
)

import { DisclosureProfile, VerificationFailure } from "@baselayer/kya";

// POST /nonces
async function nonceEndpoint(): Promise<Response> {
  return Response.json({
    nonce: await verifier.issueNonce(),
    audience: verifier.audience,
  });
}

// The same JSON you publish at /.well-known/kya-profile.json.
const profile = DisclosureProfile.parse(JSON.parse(profileJson));

// POST /oauth/token. subjectToken and scopeReference are the
// subject_token and scope fields of the form body.
async function tokenEndpoint(subjectToken: string, scopeReference: string): Promise<Response> {
  const scope = profile.resolveReference(scopeReference);
  if (scope === null) return oauthError(VerificationFailure.UnknownDisclosureScope);

  const report = await verifier.verify(subjectToken, { scope });
  if (!report.accepted) return oauthError(report.failure);

  const claims = report.exchangeClaims();
  // Your policy decides from here.
  return tokenResponse(claims);
}

from baselayer_kya import DisclosureProfile, VerificationFailure

# The same JSON you publish at /.well-known/kya-profile.json.
profile = DisclosureProfile.model_validate_json(profile_json)


# POST /nonces
async def nonce_endpoint():
    return {"nonce": await verifier.issue_nonce(), "audience": verifier.audience}, 200


# POST /oauth/token. subject_token and scope_reference are the
# subject_token and scope fields of the form body.
async def token_endpoint(subject_token: str, scope_reference: str):
    scope = profile.resolve_reference(scope_reference)
    if scope is None:
        return oauth_error(VerificationFailure.UnknownDisclosureScope)

    report = await verifier.verify(subject_token, scope=scope)
    if report.failure is not None:
        return oauth_error(report.failure)

    claims = report.exchange_claims()
    # Your policy decides from here.
    return await token_response(claims)

import { type ExchangeClaims, toOAuthError, type VerificationFailure } from "@baselayer/kya";

// The SDK picks the status and body for each failure. If Baselayer could not
// be reached, it is a retryable 503. The failure code always goes in kya_error.
function oauthError(failure: VerificationFailure): Response {
  const answer = toOAuthError(failure);
  return Response.json(answer.body(), { status: answer.status, headers: answer.headers() });
}

// Answer with your own access token, as any OAuth token route does.
async function tokenResponse(claims: ExchangeClaims, scope?: string): Promise<Response> {
  // Never let your token outlive the credential (or, on a business pair, either credential).
  const exp = claims.act !== null && claims.act.exp < claims.exp ? claims.act.exp : claims.exp;
  const expiresIn = Math.floor((exp.getTime() - Date.now()) / 1000);
  const accessToken = await issueAccessToken({ // your existing token code
    ...claims.toTokenClaims(), // sub and exp (plus act on a business pair)
    checks: claims.verification.active_keys, // the checks that passed, e.g. "phone_possession"
    scope,
  });
  return Response.json({
    access_token: accessToken,
    issued_token_type: "urn:ietf:params:oauth:token-type:access_token",
    token_type: "Bearer",
    expires_in: expiresIn,
  });
}

import time

from baselayer_kya import ExchangeClaims, VerificationFailure


# The SDK picks the status and body for each failure. If Baselayer could not
# be reached, it is a retryable 503. The failure code always goes in kya_error.
def oauth_error(failure: VerificationFailure):
    answer = failure.to_oauth_error()
    return answer.body(), answer.status


# Answer with your own access token, as any OAuth token route does.
async def token_response(claims: ExchangeClaims, scope: str | None = None):
    # Never let your token outlive the credential (or, on a business pair, either credential).
    exp = min(claims.exp, claims.act.exp) if claims.act is not None else claims.exp
    expires_in = int(exp.timestamp() - time.time())
    access_token = await issue_access_token({  # your existing token code
        **claims.to_token_claims(),  # sub and exp (plus act on a business pair)
        # The checks that passed, e.g. "phone_possession".
        "checks": list(claims.verification.active_keys),
        "scope": scope,
    })
    return {
        "access_token": access_token,
        "issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
        "token_type": "Bearer",
        "expires_in": expires_in,
    }, 200