feat(integrations): add opt-in AURA trust adapter (#2026)

Adds a read-only opt-in AURA trust-check adapter. Synthetic merge passed validators and 22 AURA offline tests via uvx pytest.
This commit is contained in:
luisllaver 2026-05-25 15:10:42 -03:00 committed by GitHub
parent c2b3899685
commit 1d72dfb2d5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 650 additions and 0 deletions

126
integrations/aura/README.md Normal file
View File

@ -0,0 +1,126 @@
# AURA trust-check adapter
Opt-in, **read-only** counterparty reputation for agent hosts. One HTTP GET
answers *"can I trust this agent before I delegate work or settle a payment?"*
- **Zero dependencies** — pure Python stdlib. Vendor the `aura/` folder, no `pip install`.
- **Read-only** — the only network call is `GET /check?did=...`. No auth, no API key.
- **No coupling** — does not sign, hold keys, move funds, or touch your wallet.
- **Off by default** — nothing runs until you call it. Disabled = delete the import.
## Enable (opt-in)
It's a gate you call explicitly at a trust boundary — there is no global hook,
no monkey-patching, no background calls. Wrap the action you want to protect:
```python
from aura import before_settle, AuraUntrusted
def settle(counterparty_did: str, amount: float) -> None:
try:
before_settle(counterparty_did) # rejects high_risk + unknown
except AuraUntrusted as e:
log.warning("blocked: %s", e)
return # your policy decides what to do
pay(counterparty_did, amount) # your existing logic, untouched
```
Prefer to read the verdict yourself instead of raising?
```python
from aura import aura_verdict
v = aura_verdict(counterparty_did)
print(v.verdict) # trusted | caution | high_risk | new | unknown
print(v.reason) # human-readable explanation
print(v.score) # composite 0..1, or None when there's no history
print(v.ok) # True for trusted/caution
# v.dimensions tells you *which* axis is weak, not just the aggregate:
if v.dimensions and v.dimensions.get("financial_integrity", 1) < 0.4:
require_manual_review() # placeholder for your own policy
```
> `v.ok` reflects the *verdict class* (True for `trusted`/`caution`), not the
> outcome of `require_trust()` — the gate's default `allow` also lets `new`
> through. Use the gate's return/raise for the decision, `v.ok` for display.
## Verdicts
| verdict | meaning | `ok` |
|---|---|---|
| `trusted` | strong on-chain track record (composite ≥ 0.70) | ✅ |
| `caution` | mixed history (0.400.70) | ✅ |
| `high_risk` | poor track record (< 0.40) | |
| `new` | registered identity, no interactions yet | ❌ |
| `unknown` | no track record — or AURA was unreachable | ❌ |
## Policy knobs
```python
# Reject brand-new agents too (strict):
before_settle(did, allow=("trusted", "caution"))
# Treat an *unreachable* AURA as a pass (fail-open). Off by default —
# absence of evidence is not evidence of trust.
before_settle(did, fail_open=True)
# Point at a self-hosted / staging gateway:
before_settle(did, base_url="https://my-aura-mirror.example", timeout=5)
```
`require_trust` is an alias of `before_settle` for non-payment call sites.
## Failure behavior
`aura_verdict()` **never raises on a network or parse error** — it returns an
`unknown` verdict with the reason set. The gate then decides:
- **default (`fail_open=False`)**`unknown` is rejected → an unreachable AURA
blocks the action. *Fail-closed.*
- **`fail_open=True`** — `unknown` from an unreachable endpoint is allowed
through, so AURA can never take your flow down. *Fail-open.*
This keeps the trust signal **purely additive**: if you remove the adapter or
AURA is down, your existing allow/deny logic runs exactly as before.
## Tests
Offline — every call replays a recorded `/check` body, no network:
```bash
python -m pytest aura/tests -q
```
Covers all five verdict classes, the gate's allow-list + `fail_open`, the
unreachable path, and input validation. See `tests/fixtures.py` for the
recorded response shapes.
## Boundary & threats
See [THREAT_MODEL.md](./THREAT_MODEL.md) — what the verdict does and does not
prove, and the failure modes a verifier should account for.
## Carry the AURA badge
Show your live trust verdict in your own README — it updates automatically and
links back to your AURA profile:
```markdown
[![AURA Verified](https://agent.auraopenprotocol.org/badge?did=YOUR_DID)](https://agent.auraopenprotocol.org/check?did=YOUR_DID)
```
A shields-style badge colored by verdict (`trusted` green, `caution` amber,
`high_risk` red, `new` blue, `unknown` grey). Add `&score=1` to show the
composite score. No DID yet? The bare badge is a generic mark:
```markdown
[![Powered by AURA](https://agent.auraopenprotocol.org/badge)](https://auraopenprotocol.org)
```
## What's behind the verdict
[AURA Open Protocol](https://auraopenprotocol.org) — W3C DID identity plus 8
on-chain reputation dimensions on Base L2 (`task_completion`, `delivery_speed`,
`output_quality`, `honesty`, `financial_integrity`, `security_compliance`,
`collaboration`, `dispute_history`). Docs: https://dev.auraopenprotocol.org

View File

@ -0,0 +1,55 @@
# Threat model — AURA trust-check adapter
A short, honest boundary statement. The verdict is **one backward-looking
signal**, not a security guarantee. Read this before treating `trusted` as a
green light for anything irreversible.
## What the verdict proves
- The DID has (or lacks) an on-chain interaction history on AURA, summarized
into a composite score and per-dimension breakdown.
- It is **backward-looking**: a statement about past recorded behavior, not a
prediction or an authorization for the *current* proposed action.
## What it explicitly does NOT prove
- **Not action-safety.** A `trusted` agent can still propose a malicious or
buggy transaction. Pair this with a forward-looking action-risk check
(contract simulation, policy engine) and keep the two signals separate so
the policy decision stays auditable.
- **Not execution quality.** It says nothing about whether *this* call will
succeed.
- **Not identity proof of the live caller.** It checks a DID's reputation, not
that the entity you're talking to controls that DID (see "Spoofed DID").
## Failure modes a caller must account for
| # | Threat | Mitigation in this adapter | Residual risk owned by caller |
|---|---|---|---|
| 1 | **Endpoint unreachable / timeout** | Returns `unknown` (never raises). Gate is fail-closed by default. | Choose `fail_open` deliberately; pick a sane `timeout`. |
| 2 | **Spoofed DID** — caller claims a DID it doesn't control | Out of scope: adapter checks reputation, not control of the key. | Verify DID control (signature challenge / auth) **before** trusting the verdict. |
| 3 | **Stale verdict** — score lags very recent bad behavior | Each call is live (no caching here). | If you cache the result, bound the TTL; don't reuse a verdict across sessions. |
| 4 | **Endpoint MITM / response tampering** | HTTPS to a pinned host (`agent.auraopenprotocol.org`). Verdict strings are validated against a fixed allow-list; unknown values collapse to `unknown`. | Don't point `base_url` at an untrusted mirror. Consider TLS pinning if your runtime supports it. |
| 5 | **Score gaming / Sybil** — cheap DIDs farming a `trusted` score | Inherited from AURA's on-chain cost + dispute dimension; not solvable in the adapter. | Weight `dimensions` (e.g. require non-trivial `interactions` / `dispute_history`) for high-value actions rather than trusting the aggregate alone. |
| 6 | **Over-trust** — using the verdict as sole gate for irreversible value | `new`/`unknown` rejected by default; `dimensions` exposed. | For high-value settlement, combine with action-risk + escrow + manual review. |
## Data handled
- **Sent:** only the counterparty DID, as a query parameter to `/check`. No
PII, no payloads, no secrets, no keys.
- **Stored:** nothing. The adapter is stateless; it holds the DID only for the
duration of the call.
- **Received:** the public `/check` JSON body. Surfaced verbatim on `.raw`.
## Trust boundary summary
```
your host --(DID only, HTTPS GET)--> AURA /check --> verdict
| |
| forward-looking action-risk check (separate, yours) |
v v
policy decision (auditable, your code)
```
The adapter sits on the read-only reputation edge. Signing, fund movement,
and the final allow/deny decision stay in your code, where they can be audited.

View File

@ -0,0 +1,36 @@
"""
AURA trust-check adapter opt-in, read-only counterparty reputation.
from aura import before_settle, AuraUntrusted
try:
before_settle(counterparty_did)
settle_payment(counterparty_did, amount)
except AuraUntrusted as e:
abort(str(e))
Zero dependencies (pure stdlib). Does not sign, hold keys, or move funds.
See README.md for the enable section and THREAT_MODEL.md for the boundary.
"""
from .adapter import (
DEFAULT_ALLOW,
DEFAULT_BASE_URL,
AuraUntrusted,
AuraVerdict,
aura_verdict,
before_settle,
require_trust,
)
__all__ = [
"aura_verdict",
"before_settle",
"require_trust",
"AuraVerdict",
"AuraUntrusted",
"DEFAULT_BASE_URL",
"DEFAULT_ALLOW",
]
__version__ = "0.1.0"

View File

@ -0,0 +1,206 @@
"""
AURA trust-check adapter a zero-dependency, read-only reputation lookup.
Drop this module into any agent/host project to gate a sensitive action
(settlement, delegation, tool execution) behind a backward-looking trust
verdict for the *counterparty* agent. It does NOT sign, hold keys, move
funds, or touch your wallet. It makes one HTTP GET and returns a verdict.
Design boundary (intentional):
- read-only: the only network call is GET /check?did=...
- no auth: /check is a public endpoint; no API key, no secret
- no coupling: pure stdlib (urllib). No third-party imports, no SDK.
- fail-closed: on network failure the verdict is `unknown`, and the
default gate (before_settle) rejects `unknown` so an
unreachable AURA never silently waves a counterparty
through. Flip `fail_open=True` to invert that.
Public API:
aura_verdict(did) -> AuraVerdict (never raises on network)
before_settle(did, allow=...) -> AuraVerdict (raises AuraUntrusted)
require_trust = before_settle (alias)
"""
from __future__ import annotations
import json
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
__all__ = [
"aura_verdict",
"before_settle",
"require_trust",
"AuraVerdict",
"AuraUntrusted",
"DEFAULT_BASE_URL",
"DEFAULT_ALLOW",
]
DEFAULT_BASE_URL = "https://agent.auraopenprotocol.org"
DEFAULT_TIMEOUT = 8 # seconds
# Verdicts safe to proceed with by default. Rejects `high_risk` (poor track
# record) and `unknown` (no verifiable history / endpoint unreachable).
DEFAULT_ALLOW = ("trusted", "caution", "new")
# All verdict classes the /check endpoint can return.
VERDICTS = ("trusted", "caution", "high_risk", "new", "unknown")
class AuraUntrusted(Exception):
"""Raised by before_settle() when a counterparty fails the trust gate."""
def __init__(self, verdict: "AuraVerdict") -> None:
self.verdict = verdict
super().__init__(
f"trust gate rejected {verdict.did}: {verdict.verdict}{verdict.reason}"
)
@dataclass(frozen=True)
class AuraVerdict:
"""
Result of a zero-auth trust check on a counterparty DID.
Fields:
did the DID that was checked
verdict one of trusted | caution | high_risk | new | unknown
reason human-readable explanation
score composite 0..1, or None when there is no history
has_history True once the agent has on-chain interactions
dimensions per-dimension breakdown (which axis is weak), or None
raw the untouched JSON body, for callers that want more
"""
did: str
verdict: str
reason: str = ""
score: Optional[float] = None
has_history: bool = False
dimensions: Optional[dict[str, float]] = None
# False only when AURA could not be reached (network/parse failure) and the
# verdict is a synthetic `unknown`. A reachable AURA that genuinely returns
# `unknown` has reachable=True. before_settle's fail_open keys on this, not
# on the verdict alone, so it can't wave through unverified counterparties.
reachable: bool = True
raw: dict[str, Any] = field(default_factory=dict, repr=False)
@property
def ok(self) -> bool:
"""True for verdicts safe to proceed with (trusted / caution)."""
return self.verdict in ("trusted", "caution")
def as_dict(self) -> dict[str, Any]:
"""The minimal {verdict, reason, score} contract, plus did/ok."""
return {
"did": self.did,
"verdict": self.verdict,
"reason": self.reason,
"score": self.score,
"ok": self.ok,
}
@classmethod
def from_payload(cls, did: str, body: dict[str, Any]) -> "AuraVerdict":
verdict = str(body.get("verdict", "unknown"))
if verdict not in VERDICTS:
verdict = "unknown"
return cls(
did=body.get("did", did),
verdict=verdict,
reason=str(body.get("reason", "")),
score=body.get("score"),
has_history=bool(body.get("has_history", False)),
dimensions=body.get("dimensions"),
raw=body,
)
@classmethod
def unreachable(cls, did: str, reason: str) -> "AuraVerdict":
"""A synthetic `unknown` verdict for network/parse failures."""
return cls(did=did, verdict="unknown", reason=reason, reachable=False)
# Indirection point so tests can inject canned responses without a network.
# Signature: (url: str, timeout: float) -> dict (raises on transport error)
def _http_get_json(url: str, timeout: float) -> dict[str, Any]:
req = urllib.request.Request(url, headers={"User-Agent": "aura-adapter/1.0"})
with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310 (https only)
return json.loads(resp.read().decode("utf-8"))
def aura_verdict(
did: str,
*,
base_url: str = DEFAULT_BASE_URL,
timeout: float = DEFAULT_TIMEOUT,
_fetch: Callable[[str, float], dict[str, Any]] = _http_get_json,
) -> AuraVerdict:
"""
Look up the trust verdict for a counterparty DID. Never raises on a
network/parse failure returns an `unknown` verdict instead, leaving the
proceed/abort decision to the caller's policy (see before_settle).
v = aura_verdict("did:aura:z6Mk...")
print(v.verdict, v.reason, v.score)
`_fetch` is an injection seam for tests; production callers ignore it.
"""
if not did or not str(did).startswith("did:"):
raise ValueError(f"invalid DID: {did!r} (must start with 'did:')")
url = f"{base_url.rstrip('/')}/check?" + urllib.parse.urlencode({"did": did})
try:
body = _fetch(url, timeout)
except (urllib.error.URLError, TimeoutError, OSError) as e:
return AuraVerdict.unreachable(did, f"AURA unreachable: {e}")
except (json.JSONDecodeError, ValueError) as e:
return AuraVerdict.unreachable(did, f"AURA returned non-JSON: {e}")
if not isinstance(body, dict):
return AuraVerdict.unreachable(did, "AURA returned an unexpected shape")
return AuraVerdict.from_payload(did, body)
def before_settle(
did: str,
*,
allow: tuple[str, ...] = DEFAULT_ALLOW,
fail_open: bool = False,
base_url: str = DEFAULT_BASE_URL,
timeout: float = DEFAULT_TIMEOUT,
_fetch: Callable[[str, float], dict[str, Any]] = _http_get_json,
) -> AuraVerdict:
"""
Gate a sensitive action behind a trust check. Returns the verdict on pass,
raises AuraUntrusted on fail.
try:
before_settle(counterparty_did) # rejects high_risk + unknown
settle_payment(counterparty_did, amount)
except AuraUntrusted as e:
abort(str(e))
Tighten to reject brand-new agents too:
before_settle(did, allow=("trusted", "caution"))
fail_open=True makes an *unreachable* AURA pass through (transport failure
only a reachable AURA that returns `unknown` is still rejected). Off by
default absence of evidence is not evidence of trust.
"""
v = aura_verdict(did, base_url=base_url, timeout=timeout, _fetch=_fetch)
if v.verdict in allow:
return v
# fail_open only excuses a transport failure, never a reachable `unknown`.
if fail_open and not v.reachable:
return v
raise AuraUntrusted(v)
# Alias — same gate, name that reads better at non-payment call sites.
require_trust = before_settle

View File

View File

@ -0,0 +1,94 @@
"""
Canned /check responses one per verdict class.
These are recorded shapes of real GET /check?did=... responses, used so the
test suite runs offline with no network. Pass `make_fetch(...)` as the
`_fetch` argument to aura_verdict / before_settle to replay them.
"""
from __future__ import annotations
from typing import Any, Callable
# did -> recorded /check JSON body
RECORDED: dict[str, dict[str, Any]] = {
"did:aura:trusted-bot": {
"did": "did:aura:trusted-bot",
"verdict": "trusted",
"reason": "strong on-chain track record (composite 0.86)",
"has_history": True,
"score": 0.86,
"interactions": 142,
"dimensions": {
"task_completion": 0.92,
"delivery_speed": 0.81,
"output_quality": 0.88,
"honesty": 0.90,
"financial_integrity": 0.95,
"security_compliance": 0.79,
"collaboration": 0.84,
"dispute_history": 0.83,
},
},
"did:aura:caution-bot": {
"did": "did:aura:caution-bot",
"verdict": "caution",
"reason": "mixed history (composite 0.55)",
"has_history": True,
"score": 0.55,
"interactions": 31,
"dimensions": {"financial_integrity": 0.41, "task_completion": 0.62},
},
"did:aura:risky-bot": {
"did": "did:aura:risky-bot",
"verdict": "high_risk",
"reason": "poor track record (composite 0.22)",
"has_history": True,
"score": 0.22,
"interactions": 18,
"dimensions": {"financial_integrity": 0.12, "dispute_history": 0.20},
},
"did:aura:fresh-bot": {
"did": "did:aura:fresh-bot",
"verdict": "new",
"reason": "registered identity, no interactions yet",
"has_history": False,
"score": None,
"interactions": 0,
},
"did:aura:ghost-bot": {
"did": "did:aura:ghost-bot",
"verdict": "unknown",
"reason": "no track record — unverified counterparty",
"has_history": False,
"score": None,
"interactions": 0,
},
}
def make_fetch(
table: dict[str, dict[str, Any]] | None = None,
) -> Callable[[str, float], dict[str, Any]]:
"""
Build a `_fetch` stand-in that replays RECORDED bodies by DID parsed from
the query string. Unknown DIDs replay the `unknown` body.
"""
table = RECORDED if table is None else table
def _fetch(url: str, timeout: float) -> dict[str, Any]:
from urllib.parse import parse_qs, urlparse
did = parse_qs(urlparse(url).query).get("did", [""])[0]
return table.get(did, RECORDED["did:aura:ghost-bot"])
return _fetch
def raising_fetch(exc: Exception) -> Callable[[str, float], dict[str, Any]]:
"""Build a `_fetch` that always raises — simulates an unreachable AURA."""
def _fetch(url: str, timeout: float) -> dict[str, Any]:
raise exc
return _fetch

View File

@ -0,0 +1,133 @@
"""
Offline tests for the AURA trust-check adapter.
Runs with plain `pytest` (or `python -m pytest`). No network: every call
replays a recorded /check body via the `_fetch` injection seam.
Coverage:
- one assertion per verdict class (trusted / caution / high_risk / new / unknown)
- the before_settle gate: allow-list pass/reject, custom allow, fail_open
- the network-failure path (fail-closed by default, pass with fail_open)
- input validation
"""
from __future__ import annotations
import urllib.error
import pytest
from aura.adapter import AuraUntrusted, aura_verdict, before_settle
from aura.tests.fixtures import make_fetch, raising_fetch
FETCH = make_fetch()
# ── verdict classes ─────────────────────────────────────────────────────────
@pytest.mark.parametrize(
"did,expected,ok",
[
("did:aura:trusted-bot", "trusted", True),
("did:aura:caution-bot", "caution", True),
("did:aura:risky-bot", "high_risk", False),
("did:aura:fresh-bot", "new", False),
("did:aura:ghost-bot", "unknown", False),
],
)
def test_verdict_classes(did, expected, ok):
v = aura_verdict(did, _fetch=FETCH)
assert v.verdict == expected
assert v.ok is ok
assert v.did == did
assert isinstance(v.reason, str) and v.reason
def test_minimal_dict_contract():
v = aura_verdict("did:aura:trusted-bot", _fetch=FETCH)
d = v.as_dict()
assert set(d) >= {"verdict", "reason", "score"}
assert d["verdict"] == "trusted"
assert d["score"] == 0.86
def test_dimensions_exposed_for_history():
v = aura_verdict("did:aura:risky-bot", _fetch=FETCH)
assert v.has_history is True
assert v.dimensions["financial_integrity"] == 0.12
def test_new_agent_has_no_score():
v = aura_verdict("did:aura:fresh-bot", _fetch=FETCH)
assert v.score is None
assert v.has_history is False
# ── the before_settle gate ───────────────────────────────────────────────────
def test_gate_allows_trusted():
v = before_settle("did:aura:trusted-bot", _fetch=FETCH)
assert v.verdict == "trusted"
def test_gate_allows_caution_and_new_by_default():
assert before_settle("did:aura:caution-bot", _fetch=FETCH).verdict == "caution"
assert before_settle("did:aura:fresh-bot", _fetch=FETCH).verdict == "new"
def test_gate_rejects_high_risk():
with pytest.raises(AuraUntrusted) as ei:
before_settle("did:aura:risky-bot", _fetch=FETCH)
assert ei.value.verdict.verdict == "high_risk"
def test_gate_rejects_unknown_by_default():
with pytest.raises(AuraUntrusted):
before_settle("did:aura:ghost-bot", _fetch=FETCH)
def test_strict_allow_rejects_new():
with pytest.raises(AuraUntrusted):
before_settle("did:aura:fresh-bot", allow=("trusted", "caution"), _fetch=FETCH)
# ── network-failure path ──────────────────────────────────────────────────────
def test_unreachable_returns_unknown_not_raise():
fetch = raising_fetch(urllib.error.URLError("connection refused"))
v = aura_verdict("did:aura:trusted-bot", _fetch=fetch)
assert v.verdict == "unknown"
assert "unreachable" in v.reason.lower()
def test_gate_fail_closed_on_unreachable():
fetch = raising_fetch(urllib.error.URLError("connection refused"))
with pytest.raises(AuraUntrusted):
before_settle("did:aura:trusted-bot", _fetch=fetch)
def test_gate_fail_open_passes_on_unreachable():
fetch = raising_fetch(urllib.error.URLError("connection refused"))
v = before_settle("did:aura:trusted-bot", fail_open=True, _fetch=fetch)
assert v.verdict == "unknown"
assert v.reachable is False
def test_fail_open_does_not_pass_reachable_unknown():
# A reachable AURA that returns `unknown` (ghost DID) is still rejected even
# with fail_open — fail_open only excuses transport failures.
with pytest.raises(AuraUntrusted):
before_settle("did:aura:ghost-bot", fail_open=True, _fetch=FETCH)
def test_reachable_verdict_marked_reachable():
v = aura_verdict("did:aura:ghost-bot", _fetch=FETCH)
assert v.reachable is True
# ── input validation ──────────────────────────────────────────────────────────
@pytest.mark.parametrize("bad", ["", "not-a-did", "z6Mk-no-prefix", None])
def test_rejects_bad_did(bad):
with pytest.raises(ValueError):
aura_verdict(bad, _fetch=FETCH)