PyBook 12: The HTTP API

from fastapi.testclient import TestClient

from auradefi.api.app import create_app
from auradefi.api.deps import Deps
from auradefi.chains.registry import ChainRegistry
from auradefi.clock import FrozenClock
from auradefi.ledger.backends.memory import MemoryLedger
from auradefi.tenancy.audit import AuditLog
from auradefi.tenancy.keys import ApiKeyStore
from auradefi.tenancy.models import Environment, Scope
from auradefi.tenancy.quota import QuotaCounter, QuotaLimits
from auradefi.tenancy.store import TenancyStore
from auradefi.tenancy.tokens import RevocationSet
from auradefi.webhooks.deliver import WebhookStore

NOW = 1_754_000_000_000
clock = FrozenClock(NOW)

tenancy = TenancyStore()
organisation = tenancy.create_organisation("acme", clock)
project = tenancy.create_project(organisation.id, "main", Environment.LIVE, clock)
secrets_vault = {project.id: project.signing_secret}
webhooks = WebhookStore()
chains = ChainRegistry()

deps = Deps(
    tenancy=tenancy,
    keys=ApiKeyStore(),
    quota=QuotaCounter(QuotaLimits(1_000, 10_000, 100_000), clock),
    audit=AuditLog(),
    revocations=RevocationSet(),
    ledger=MemoryLedger(),
    webhooks=webhooks,
    chains=chains,
    clock=clock,
    signing_secret_for=secrets_vault.get,
    capabilities={"eip155:1": frozenset({"balances", "transactions", "prices"})},
)
client = TestClient(create_app(deps))

schema = client.get("/openapi.json").json()
assert schema["info"]["title"] == "auradefi"
assert "/batch/holdings" not in schema["paths"]   # unbound capability => no route at all
for path in sorted(schema["paths"]):
    for method in sorted(schema["paths"][path]):
        print(f"{method.upper():<6} {path}")

POST   /auth/revoke
POST   /auth/token
GET    /connections
POST   /connections
GET    /connections/{connection_id}
GET    /coverage
GET    /crypto/sync
GET    /users
GET    /users/me
GET    /webhooks/dead_letter
GET    /webhooks/deliveries
POST   /webhooks/deliveries/{delivery_id}/replay
GET    /webhooks/endpoints
POST   /webhooks/endpoints
/opt/hostedtoolcache/Python/3.12.13/x64/lib/python3.12/site-packages/fastapi/testclient.py:1: StarletteDeprecationWarning: Using `httpx` with `starlette.testclient` is deprecated; install `httpx2` instead.
  from starlette.testclient import TestClient as TestClient  # noqa

key, plaintext = deps.keys.issue(
    project.id, Environment.LIVE,
    (Scope.USERS_ADMIN, Scope.ACCOUNTS_READ, Scope.ACCOUNTS_WRITE), clock,
)
assert plaintext.startswith("adk_live_") and len(plaintext) == 57
key_headers = {"Authorization": f"Bearer {plaintext}"}

minted = client.post(
    "/auth/token",
    json={"external_user_id": "host-user-7"},
    headers={**key_headers, "X-Forwarded-For": "198.51.100.9"},
)
assert minted.status_code == 200
assert list(minted.json()) == ["token"]          # exactly one key: nothing leaks
token_headers = {"Authorization": f"Bearer {minted.json()['token']}"}

(entry,) = deps.audit.entries(project.id)
# The audited IP is the SOCKET PEER, not the X-Forwarded-For the request
# carries: trusted_proxy_hops defaults to 0, so no proxy is trusted and a
# caller cannot choose the address its own permanent audit row records
# (0.1.1 #30). entry.ip_source says which source the value came from.
assert (entry.seq, entry.event, entry.key_id, entry.ip) == (1, "token.minted", key.id, "testclient")

limits = {name.lower(): value for name, value in minted.headers.items()
          if name.lower().startswith("x-ratelimit")}
assert len(limits) == 9                       # 3 windows x limit/remaining/reset
assert limits["x-ratelimit-limit-second"] == "1000"
assert limits["x-ratelimit-remaining-second"] == "999"
print("audited:", entry.event, "by", entry.key_id, "from", entry.ip)
print("rate limit headers:", {k: v for k, v in sorted(limits.items()) if k.endswith("second")})

audited: token.minted by key_10d096175c4562dc from testclient
rate limit headers: {'x-ratelimit-limit-second': '1000', 'x-ratelimit-remaining-second': '999', 'x-ratelimit-reset-second': '1754000001000'}

ADDRESS = "0xAAAAaaaaAAAAaaaaAAAAaaaaAAAAaaaaAAAAaaaa"

created = client.post("/connections", json={"kind": "address", "descriptor": ADDRESS}, headers=token_headers)
assert created.status_code == 201
connection_id = created.json()["id"]
assert connection_id.startswith("conn_")

conflict = client.post(
    "/connections", json={"kind": "address", "descriptor": ADDRESS.lower()}, headers=token_headers
)
assert conflict.status_code == 409
assert conflict.json()["error"]["existing_connection_id"] == connection_id
assert conflict.json()["error"]["type"] == "ConflictError"

me = client.get("/users/me", headers=token_headers).json()
assert me["id"].startswith("usr_")
print("created", connection_id, "| repost ->", conflict.status_code, conflict.json()["error"]["message"])
print("caller:", me)

created conn_b955659e2e55d4ca | repost -> 409 connection already exists: 'conn_b955659e2e55d4ca'
caller: {'id': 'usr_7f859501dc340dcd', 'project_id': 'proj_08ac97861ac0e345', 'external_user_id': 'host-user-7', 'created_at_ms': 1754000000000}

from auradefi.ledger.models import Direction, Entry, LedgerTransaction, transaction_id
from auradefi.money.quantity import Quantity
from auradefi.tenancy.models import end_user_id

CHAIN, ASSET = "eip155:1", "eip155:1/slip44:60"
tenant = end_user_id(project.id, "host-user-7")
assert tenant == me["id"]


def txn(index):
    tx_hash = "0x" + f"{index:02x}" * 32
    return LedgerTransaction(
        id=transaction_id(CHAIN, tx_hash, "acct_eth"), chain_id=CHAIN, tx_hash=tx_hash,
        account_id="acct_eth", block_number=18_000_000 + index,
        initiated_at=1_753_000_000_000 + index, confirmed_at=1_753_000_000_500 + index,
        entries=(Entry(asset_id=ASSET, quantity=Quantity(index * 10**17, 18), direction=Direction.IN),),
    )


deps.ledger.upsert(tenant, [txn(index) for index in (1, 2, 3)])

seen, cursor, pages = [], None, 0
while True:
    query = "/crypto/sync?limit=2" + (f"&cursor={cursor}" if cursor else "")
    page = client.get(query, headers=token_headers).json()
    pages += 1
    assert set(page) == {"added", "modified", "removed", "next_cursor", "has_more"}
    assert page["modified"] == [] and page["removed"] == []
    seen.extend(row["transaction_id"] for row in page["added"])
    cursor = page["next_cursor"]
    if not page["has_more"]:
        break

assert pages == 2 and len(seen) == 3
assert cursor == "00000000000000000003"
sample = page["added"][0]["entries"][0]["quantity"]
assert isinstance(sample["raw"], str)               # rule #2, at the boundary
assert sample == {"raw": "300000000000000000", "decimals": 18,
                  "numeric": "0.3", "float": 0.3}
print(f"{pages} pages, {len(seen)} transactions, final cursor {cursor}")
print("wire quantity:", sample)

2 pages, 3 transactions, final cursor 00000000000000000003
wire quantity: {'raw': '300000000000000000', 'decimals': 18, 'numeric': '0.3', 'float': 0.3}

coverage = client.get("/coverage").json()   # no auth: it is a public fact
assert coverage["capabilities"] == ["balances", "transactions", "positions", "prices", "xpub"]

rows = {row["chain_id"]: row for row in coverage["chains"]}
assert rows["eip155:1"]["capabilities"] == {
    "balances": True, "transactions": True, "positions": False, "prices": True, "xpub": False,
}
assert rows["bip122:000000000019d6689c085ae165831e93"]["capabilities"] == {
    "balances": False, "transactions": False, "positions": False, "prices": False, "xpub": False,
}
print(f"{'chain':<40} {'name':<10} " + " ".join(f"{c:<12}" for c in coverage["capabilities"]))
for row in coverage["chains"]:
    flags = " ".join(f"{str(row['capabilities'][c]):<12}" for c in coverage["capabilities"])
    print(f"{row['chain_id']:<40} {row['name']:<10} {flags}")

chain                                    name       balances     transactions positions    prices       xpub        
bip122:000000000019d6689c085ae165831e93  Bitcoin    False        False        False        False        False       
eip155:1                                 Ethereum   True         True         False        True         False       
eip155:137                               Polygon    False        False        False        False        False       
eip155:8453                              Base       False        False        False        False        False       
solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp  Solana     False        False        False        False        False

import httpx

from auradefi.webhooks.deliver import Deliverer
from auradefi.webhooks.sign import verify_signature

HOOK_URL = "https://hooks.example.com/inbox"

registered = client.post("/webhooks/endpoints", json={"url": HOOK_URL}, headers=key_headers)
assert registered.status_code == 201
secret = registered.json()["secret"]
assert len(secret) == 64
listed = client.get("/webhooks/endpoints", headers=key_headers).json()["endpoints"][0]
assert "secret" not in listed          # shown once, at creation, and never again
print("endpoint", listed["id"], "registered; secret shown once")

clock.advance(1_000)
queued_at = clock.now_ms()
client.post("/connections", json={"kind": "address", "descriptor": "0x" + "bb" * 20}, headers=token_headers)

pending = client.get("/webhooks/deliveries", headers=key_headers).json()
assert pending["count"] == 1
assert pending["deliveries"][0]["event_name"] == "connection.created"
assert pending["deliveries"][0]["status"] == "pending"
print("queued:", pending["deliveries"][0]["event_name"], "at", pending["deliveries"][0]["next_attempt_at_ms"])

endpoint whe_4968735032958cee registered; secret shown once
queued: connection.created at 1754000001000

class Recorder:
    "A MockTransport handler: records every request, answers one status."

    def __init__(self, status_code):
        self.status_code, self.requests = status_code, []

    def __call__(self, request):
        self.requests.append(request)
        return httpx.Response(self.status_code)


recorder = Recorder(200)
deliverer = Deliverer(webhooks, httpx.Client(transport=httpx.MockTransport(recorder)))
(delivered,) = deliverer.tick(queued_at)

assert len(recorder.requests) == 1
sent = recorder.requests[0]
assert sent.method == "POST" and str(sent.url) == HOOK_URL
assert sent.headers["X-Auradefi-Timestamp"] == str(queued_at)

# The receiver's side of the contract, run for real:
verify_signature(secret, queued_at, sent.content.decode("utf-8"),
                 sent.headers["X-Auradefi-Signature"], queued_at)
assert delivered.status.value == "delivered"
assert deliverer.tick(queued_at) == ()      # a delivered row is never re-sent

print("signature:", sent.headers["X-Auradefi-Signature"][:32] + "…", "-> verified")
print("body:", sent.content.decode("utf-8")[:110] + "…")

signature: v1=7d5c1e044897d58af7620455225f1… -> verified
body: {"created_at_ms":1754000001000,"data":{"connection_id":"conn_1b93ae01f06362c6","descriptor":"0xbbbbbbbbbbbbbbb…

from auradefi.webhooks.models import RETRY_SCHEDULE_MS

# A fresh event against a receiver that is down: the pinned schedule, then
# the dead letter queue.
clock.advance(1_000)
born_at = clock.now_ms()
client.post("/connections", json={"kind": "address", "descriptor": "0x" + "cc" * 20}, headers=token_headers)

failing_recorder = Recorder(500)
failing = Deliverer(webhooks, httpx.Client(transport=httpx.MockTransport(failing_recorder)))
for attempt, offset in enumerate(RETRY_SCHEDULE_MS):
    (row,) = failing.tick(born_at + offset)
    assert row.attempts == attempt + 1 and row.last_status_code == 500

assert len(failing_recorder.requests) == 6
assert row.status.value == "dead_letter" and row.next_attempt_at_ms is None
letters = client.get("/webhooks/dead_letter", headers=key_headers).json()
assert letters["count"] == 1 and letters["deliveries"][0]["attempts"] == 6
print("retry schedule (ms):", RETRY_SCHEDULE_MS, "-> dead letter after", row.attempts, "attempts")

# Replay: a NEW delivery row, the original untouched.
clock.advance(1_000)
replayed = client.post(f"/webhooks/deliveries/{row.id}/replay", headers=key_headers)
assert replayed.status_code == 202
assert replayed.json()["id"] != row.id and replayed.json()["replay_ordinal"] == 1
assert replayed.json()["status"] == "pending"

healthy_recorder = Recorder(200)
(settled,) = Deliverer(webhooks, httpx.Client(transport=httpx.MockTransport(healthy_recorder))).tick(clock.now_ms())
assert settled.id == replayed.json()["id"] and settled.status.value == "delivered"
assert [d["status"] for d in client.get("/webhooks/deliveries", headers=key_headers).json()["deliveries"]] == [
    "delivered", "dead_letter", "delivered",
]
print("replayed", row.id, "->", settled.id, settled.status.value)

retry schedule (ms): (0, 60000, 300000, 1800000, 7200000, 86400000) -> dead letter after 6 attempts
replayed dlv_8de104ed55d604a9 -> dlv_bdf1d4ab68e8fc96 delivered

cases = [
    ("no credential", client.get("/crypto/sync")),
    ("user token where a key is required", client.get("/webhooks/endpoints", headers=token_headers)),
    ("key where a user token is required", client.get("/crypto/sync", headers=key_headers)),
    ("malformed body", client.post("/connections", json={"kind": "address"}, headers=token_headers)),
    ("bad cursor", client.get("/crypto/sync?cursor=nonsense", headers=token_headers)),
]
for label, response in cases:
    body = response.json()
    assert set(body) == {"error"}
    assert body["error"]["status"] == response.status_code
    print(f"{response.status_code}  {label:<36} {body['error']['type']}: {body['error']['message'][:52]}")

401  no credential                        AuthError: token failed authentication
401  user token where a key is required   AuthError: api key failed authentication
401  key where a user token is required   AuthError: token failed authentication
422  malformed body                       ValidationError: request validation failed
422  bad cursor                           CursorError: cursor token must be exactly 20 ASCII digits: 'nonse