Skip to content

Latest commit

 

History

History
467 lines (354 loc) · 17.5 KB

File metadata and controls

467 lines (354 loc) · 17.5 KB

DID WBA Authentication Integration Guide

中文版

This guide explains how to integrate DID WBA authentication into a Python HTTP server using the ANP SDK (anp package). After reading this guide, you can quickly add decentralized identity verification to any Python HTTP service.

1. Authentication Principles

1.1 What is DID WBA

DID WBA (Web-Based Agent) is a decentralized identity authentication method based on the W3C DID standard. It allows clients to include a DID and signature in the first HTTP request, enabling servers to verify client identity without additional round-trips.

Key characteristics:

  • Asymmetric cryptography: client holds private key, server obtains public key from the DID document to verify signatures
  • The current SDK defaults to HTTP Message Signatures; the old Authorization: DIDWba ... header remains available for compatibility
  • After initial authentication, the server issues an access token; the standard response header is Authentication-Info, with a compatibility Authorization: Bearer ... header during migration
  • Works with HTTPS: client verifies server via TLS certificate, server verifies client via DID signature

Important current defaults

  • create_did_wba_document() creates e1_ path-based DIDs by default
  • DIDWbaAuthHeader emits Signature-Input / Signature by default
  • resolve_did_wba_document() always validates e1_ / k1_ DID binding
  • for e1_, DID Document proof is part of binding validation and is therefore required

1.2 Authentication Flow

First Request (default HTTP Message Signatures):
Client                              Server                    Client DID Server
  |                                    |                            |
  |-- HTTP Request ------------------>|                            |
  |   Signature-Input: sig1=(...)     |                            |
  |   Signature: sig1=:...:           |-- GET /user/alice/...----->|
  |   Content-Digest: sha-256=:...:   |<-- DID Document -----------|
  |                                   |                            |
  |                                   |  1. Verify DID document id / binding / proof
  |                                   |  2. Verify created / expires / nonce
  |                                   |  3. Verify HTTP Message Signature
  |                                   |  4. Verify Content-Digest
  |                                   |  5. Valid → generate JWT
  |                                   |
  |<-- HTTP Response -----------------|
  |    Authentication-Info: access_token="..." |
  |    Authorization: Bearer <JWT>    |  (compatibility)
  |                                   |

Subsequent Requests (using JWT, faster):
Client                              Server
  |                                    |
  |-- HTTP Request ------------------>|
  |   Authorization: Bearer <JWT>     |
  |                                   |  Verify JWT signature & expiry
  |<-- HTTP Response -----------------|

1.3 Default Request Format (HTTP Message Signatures)

By default, the client sends the following headers:

POST /orders HTTP/1.1
Host: api.example.com
Content-Type: application/json
Content-Digest: sha-256=:BASE64_SHA256_DIGEST:
Signature-Input: sig1=("@method" "@target-uri" "@authority" "content-digest");created=1733402096;expires=1733402156;nonce="abc123";keyid="did:wba:example.com:user:alice:e1_<fingerprint>#key-1"
Signature: sig1=:BASE64_SIGNATURE:

Key fields:

  • keyid: full DID URL of the signing verification method
  • created / expires: signature validity window
  • nonce: anti-replay value
  • Signature: signature over the RFC 9421 signature base
  • Content-Digest: integrity binding for the request body

1.3.1 Compatibility Format (Legacy DIDWba Header)

If the client explicitly uses auth_mode="legacy_didwba", or when interoperating with old clients, it can still send:

Authorization: DIDWba did="did:wba:example.com:user:alice:k1_<fingerprint>", nonce="abc123", timestamp="2024-12-05T12:34:56Z", verification_method="key-1", signature="base64url(signature_value)"

1.4 Signature Verification Process

In the default HTTP Message Signatures flow, the server:

  1. Parses Signature-Input / Signature / Content-Digest
  2. Verifies the time window and nonce
  3. Resolves the DID document based on keyid
  4. Verifies the DID document id
  5. Verifies e1_ / k1_ binding against the DID document public key
  6. For e1_, enforces DID Document proof validation and requires the proof key to be the binding key
  7. Verifies the request signature using the matching DID document public key
  8. On success, issues an RS256 JWT token

2. Installation

pip install anp
# or using uv
uv add anp

For FastAPI support:

pip install anp[api]
# or
uv add anp --extra api

3. Server-Side Integration

3.1 Core Classes

The ANP SDK provides the following classes for server-side authentication:

Class Module Purpose
DidWbaVerifierConfig anp.authentication Verifier configuration (JWT keys, expiry, etc.)
DidWbaVerifier anp.authentication Core verifier for DID WBA headers and Bearer Tokens
auth_middleware anp.openanp.middleware FastAPI middleware that intercepts requests for authentication
create_auth_middleware anp.openanp.middleware Middleware factory for one-line setup

3.2 Minimal Example: FastAPI + DID WBA Authentication

"""Minimal example of adding DID WBA authentication to a FastAPI server."""

from fastapi import FastAPI, Request
from anp.authentication import DidWbaVerifier, DidWbaVerifierConfig
from anp.openanp.middleware import auth_middleware

# 1. Prepare JWT keys (RS256)
#    Generate with: openssl genrsa -out private.pem 2048
#                   openssl rsa -in private.pem -pubout -out public.pem
with open("private.pem") as f:
    jwt_private_key = f.read()
with open("public.pem") as f:
    jwt_public_key = f.read()

# 2. Create verifier
config = DidWbaVerifierConfig(
    jwt_private_key=jwt_private_key,
    jwt_public_key=jwt_public_key,
    jwt_algorithm="RS256",
    access_token_expire_minutes=60,
)
verifier = DidWbaVerifier(config)

# 3. Create FastAPI app and register middleware
app = FastAPI()

# Paths that bypass authentication
exempt_paths = ["/health", "/docs", "/openapi.json"]

@app.middleware("http")
async def did_wba_auth(request: Request, call_next):
    return await auth_middleware(
        request, call_next, verifier, exempt_paths=exempt_paths
    )

# 4. Define routes
@app.get("/health")
async def health():
    """Exempt endpoint, no authentication required."""
    return {"status": "healthy"}

@app.get("/api/protected")
async def protected(request: Request):
    """Protected endpoint, requires DID authentication."""
    did = request.state.did          # DID is auto-injected after authentication
    return {"message": "Authentication successful", "did": did}

3.3 DidWbaVerifierConfig Reference

from anp.authentication import DidWbaVerifierConfig

config = DidWbaVerifierConfig(
    # JWT signing keys (PEM strings) for issuing and verifying Bearer Tokens
    jwt_private_key="-----BEGIN RSA PRIVATE KEY-----\n...",
    jwt_public_key="-----BEGIN PUBLIC KEY-----\n...",
    jwt_algorithm="RS256",            # JWT signing algorithm

    # Token lifetime
    access_token_expire_minutes=60,   # Bearer Token expiry in minutes

    # Security parameters
    nonce_expiration_minutes=6,       # Nonce record expiry (should be slightly > timestamp expiry)
    timestamp_expiration_minutes=5,   # Timestamp validity window in minutes

    # Optional: custom nonce validator (for distributed deployments sharing nonce state)
    # Signature: (did: str, nonce: str) -> bool (also supports async)
    external_nonce_validator=None,

    # Optional: domain whitelist to restrict authentication sources
    allowed_domains=["example.com", "localhost"],
)

3.4 How the Middleware Works

The auth_middleware workflow:

  1. Check if the request path is in exempt_paths (supports wildcards like */ad.json, /info/*)
  2. If exempt, pass through directly
  3. If authentication is required, extract the Authorization header
  4. If the request contains Signature-Input / Signature, run the default HTTP Message Signatures verification flow; if it starts with DIDWba, run the compatibility verification flow. On success:
    • Store auth result in request.state.auth_result
    • Store DID in request.state.did
    • Return the access token in the response Authentication-Info header
    • During migration, also return Authorization: Bearer <JWT> for old clients
  5. If it starts with Bearer, verify the JWT Token
  6. On authentication failure, return a 401 or 403 JSON response

3.5 Shorthand: Using create_auth_middleware

from fastapi import FastAPI
from anp.authentication import DidWbaVerifierConfig
from anp.openanp.middleware import create_auth_middleware

app = FastAPI()

config = DidWbaVerifierConfig(
    jwt_private_key=open("private.pem").read(),
    jwt_public_key=open("public.pem").read(),
    jwt_algorithm="RS256",
)

# One-line middleware registration
app.middleware("http")(
    create_auth_middleware(config, exempt_paths=["/health", "/docs"])
)

3.6 Accessing Authentication Data in Route Handlers

After middleware authentication succeeds, access auth data via request.state:

@app.get("/api/data")
async def get_data(request: Request):
    # Get the authenticated DID
    did: str = request.state.did
    # e.g., did = "did:wba:example.com:user:alice"

    # Get the full authentication result
    auth_result: dict = request.state.auth_result
    # First auth: {"access_token": "...", "token_type": "bearer", "did": "...", "auth_scheme": "...", "response_headers": {...}}
    # Bearer Token auth: {"did": "...", "auth_scheme": "bearer", "response_headers": {}}

    return {"did": did, "data": "..."}

3.7 Without Middleware: Using the Verifier Directly

If you're not using FastAPI or need finer-grained control, use DidWbaVerifier directly:

from anp.authentication import DidWbaVerifier, DidWbaVerifierConfig, DidWbaVerifierError

config = DidWbaVerifierConfig(
    jwt_private_key=open("private.pem").read(),
    jwt_public_key=open("public.pem").read(),
)
verifier = DidWbaVerifier(config)

# Use in any HTTP framework
async def handle_request(method: str, url: str, headers: dict, body: bytes):
    try:
        result = await verifier.verify_request(
            method=method,
            url=url,
            headers=headers,
            body=body,
        )
        did = result["did"]
        access_token = result.get("access_token")  # Only present for first DID auth
        return did, access_token
    except DidWbaVerifierError as e:
        # e.status_code: 401 (auth failed), 403 (forbidden), 500 (internal error)
        return None, None

This works with aiohttp, Flask, Django, or any other Python HTTP framework.

4. Client-Side Integration

4.1 Using DIDWbaAuthHeader

The SDK provides DIDWbaAuthHeader to automatically manage auth headers and token caching:

from anp.authentication import DIDWbaAuthHeader
import httpx

# 1. Create authentication client
authenticator = DIDWbaAuthHeader(
    did_document_path="path/to/did.json",       # Path to DID document
    private_key_path="path/to/private-key.pem",  # Corresponding private key
)

server_url = "https://example.com"
data_url = f"{server_url}/api/data"
other_url = f"{server_url}/api/other"

# 2. First request: auto-generates HTTP Message Signatures by default
headers = authenticator.get_auth_header(
    data_url,
    force_new=True,
    method="GET",
)
# Default: {"Signature-Input": "...", "Signature": "..."}
# With auth_mode="legacy_didwba": {"Authorization": "DIDWba ..."}

with httpx.Client() as client:
    response = client.get(data_url, headers=headers)

    # 3. Extract and cache Bearer Token from response
    authenticator.update_token(data_url, dict(response.headers))

    # 4. Subsequent requests: automatically uses cached Bearer Token
    headers = authenticator.get_auth_header(other_url)
    # headers = {"Authorization": "Bearer eyJ..."}
    response = client.get(other_url, headers=headers)

4.2 DIDWbaAuthHeader Key Methods

Method Description
get_auth_header(server_url, force_new=False, method="GET", headers=None, body=None) Get auth header. Returns a cached Bearer header if available; otherwise generates HTTP Message Signatures by default. server_url must be the final request URL (including path/query), and body must be the exact outbound bytes
get_challenge_auth_header(server_url, response_headers, method="GET", headers=None, body=None) Rebuild auth headers from a 401 response using WWW-Authenticate / Accept-Signature, automatically reusing the server-provided nonce
update_token(server_url, headers) Extract Bearer Token from response headers and cache it
clear_token(server_url) Clear cached token for the specified domain
clear_all_tokens() Clear all cached tokens

4.3 Handling Token Expiry

data_url = f"{server_url}/api/data"
response = client.get(data_url, headers=headers)

if response.status_code == 401:
    # Token expired, nonce became invalid, or the server returned a fresh challenge
    authenticator.clear_token(data_url)
    headers = authenticator.get_challenge_auth_header(
        data_url,
        dict(response.headers),
        method="GET",
    )
    response = client.get(data_url, headers=headers)

    # Cache the new token
    authenticator.update_token(data_url, dict(response.headers))

If you use the higher-level ANPClient, the Bearer refresh and 401 + WWW-Authenticate/nonce retry flow is handled automatically.

5. DID Document and Key Preparation

5.1 Generate a DID Document

Use the SDK function to generate a DID document:

from anp.authentication import create_did_wba_document

did_document, private_keys = create_did_wba_document(
    hostname="example.com",
    path_segments=["user", "alice"],
)
# Default path DID shape: did:wba:example.com:user:alice:e1_<fingerprint>
# private_keys contains the generated private key objects

import json
with open("did.json", "w") as f:
    json.dump(did_document, f, indent=2)

Or use the command-line tool:

uv run python tools/did_generater/generate_did_doc.py "did:wba:example.com:user:alice"

5.2 Generate JWT RS256 Key Pair

The server needs an RS256 key pair for issuing and verifying Bearer Tokens (separate from DID document keys):

# Generate RSA private key
openssl genrsa -out RS256-private.pem 2048

# Export public key
openssl rsa -in RS256-private.pem -pubout -out RS256-public.pem

5.3 Deploy the DID Document

The DID document must be hosted at an HTTPS-accessible path:

did:wba:example.com              → https://example.com/.well-known/did.json
did:wba:example.com:user:alice   → https://example.com/user/alice/did.json
did:wba:example.com%3A3000       → https://example.com:3000/.well-known/did.json

During verification, the server automatically resolves the corresponding URL based on the client's DID to fetch the document.

6. Runnable Examples

This directory provides complete runnable examples:

  • http_server.py — FastAPI server with DID WBA authentication middleware
  • http_client.py — HTTP client demonstrating the full authentication and token reuse flow
  • authenticate_and_verify.py — Offline demo, no server needed, shows the complete auth + verification flow
  • create_did_document.py — DID document generation example

To run:

# Terminal 1: Start the server
uv run python examples/python/did_wba_examples/http_server.py

# Terminal 2: Run the client
uv run python examples/python/did_wba_examples/http_client.py

7. Custom DID Document Resolution

By default, the SDK resolves DID documents via HTTPS. For development/testing, you can replace this with a local resolver:

from anp.authentication import did_wba_verifier as verifier_module

# Option 1: Replace the global resolution function (for testing)
async def local_resolver(did: str) -> dict:
    # Load DID document from local file or database
    return load_did_from_db(did)

verifier_module.resolve_did_wba_document = local_resolver

# Option 2: In production, just ensure the client's DID HTTPS path is accessible
# The SDK automatically fetches from https://<domain>/<path>/did.json

8. Security Considerations

  1. Private key management: Both DID private keys and JWT private keys must be stored securely. Never commit them to source control.
  2. HTTPS: Production environments must use HTTPS. DID document retrieval must also use HTTPS.
  3. Nonce anti-replay: The built-in nonce validation is in-memory (single-process only). For distributed deployments, use external_nonce_validator to integrate Redis or another shared store.
  4. Clock skew: The clock difference between server and client should not exceed timestamp_expiration_minutes.
  5. Token lifetime: Adjust access_token_expire_minutes to suit your needs. Recommended: no more than 24 hours.
  6. Domain whitelist: In production, configure allowed_domains to restrict authentication sources.

9. References