Spaces:
Build error
Build error
File size: 5,353 Bytes
690acd1 da70460 690acd1 da70460 690acd1 da70460 690acd1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | """Routes for on-chain attestation rationale.
`RiskAttestationRegistry` (arf-onchain) anchors only a `rationale_hash` --
never the reasoning itself, to keep operational detail about a customer's
infrastructure off a public chain. These endpoints are the off-chain half:
a reference risk evaluator (`arf_enterprise.onchain.evaluator`) persists the
plaintext here immediately after signing an attestation, keyed by the same
hash it put on-chain, and an auditor who reads a `DecisionAnchored` or
`AttestationIssued` event can fetch the reasoning behind it here.
Internal-key gated like `routes_governance.py`: this is a service-to-service
surface for the evaluator process and for auditor tooling, not a
tenant-scoped customer endpoint -- see `OnchainRationaleDB`'s docstring for
why there is no tenant_id to enforce here.
"""
import logging
from fastapi import APIRouter, Depends, HTTPException, Response
from pydantic import BaseModel, field_validator
from sqlalchemy.orm import Session
from app.api.deps import get_db, verify_internal_key
from app.database.models_onchain import OnchainRationaleDB
logger = logging.getLogger(__name__)
router = APIRouter(dependencies=[Depends(verify_internal_key)])
def _validate_hex_hash(value: str) -> str:
text = value.strip()
if not text.startswith("0x") or len(text) != 66:
raise ValueError("rationale_hash must be a 0x-prefixed 32-byte hex string")
try:
int(text, 16)
except ValueError:
raise ValueError("rationale_hash is not valid hex") from None
return text.lower()
class RationaleRequest(BaseModel):
rationale_hash: str
rationale: str
agent_address: str | None = None
evaluator_address: str | None = None
@field_validator("rationale_hash")
@classmethod
def _validate_hash(cls, value: str) -> str:
return _validate_hex_hash(value)
@field_validator("rationale")
@classmethod
def _validate_rationale(cls, value: str) -> str:
if not value.strip():
raise ValueError("rationale must not be empty")
return value
class RationaleResponse(BaseModel):
rationale_hash: str
rationale: str
agent_address: str | None
evaluator_address: str | None
@router.post("/onchain/rationale", status_code=201)
async def persist_rationale(
req: RationaleRequest,
response: Response,
db: Session = Depends(get_db),
):
"""Store the plaintext behind an anchored `rationale_hash`.
Idempotent on `rationale_hash`: signing the same decision twice (a
retry after a network error, for instance) posts the same hash and
text, so the second call is a no-op rather than a uniqueness-constraint
error. A *different* text arriving for a hash already on record is
refused -- that would mean either hash collision or a caller bug, and
silently overwriting an anchored record's preimage is the one thing
this table must never do.
"""
existing = (
db.query(OnchainRationaleDB)
.filter(OnchainRationaleDB.rationale_hash == req.rationale_hash)
.one_or_none()
)
if existing is not None:
if existing.rationale != req.rationale:
raise HTTPException(
status_code=409,
detail=(
"rationale_hash already recorded with different text; "
"an anchored hash's preimage cannot be overwritten"
),
)
# The route decorator's status_code=201 is FastAPI's default for
# every plain-dict return from this handler, including this one --
# it must be overridden explicitly here or a replayed post reports
# itself as newly Created.
response.status_code = 200
return {"status": "already_recorded", "rationale_hash": req.rationale_hash}
row = OnchainRationaleDB(
rationale_hash=req.rationale_hash,
rationale=req.rationale,
agent_address=req.agent_address,
evaluator_address=req.evaluator_address,
)
db.add(row)
db.commit()
logger.info("persisted rationale for hash %s", req.rationale_hash)
return {"status": "recorded", "rationale_hash": req.rationale_hash}
@router.get("/onchain/rationale/{rationale_hash}", response_model=RationaleResponse)
async def get_rationale(
rationale_hash: str,
db: Session = Depends(get_db),
):
"""Fetch the plaintext behind an anchored `rationale_hash`.
What an auditor calls after reading a `DecisionAnchored` event off-chain
-- the hash from the event is the only key this endpoint accepts, by
design: there is no listing or search here, only lookup by the exact
value that was signed and anchored.
"""
try:
normalized = _validate_hex_hash(rationale_hash)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
row = (
db.query(OnchainRationaleDB)
.filter(OnchainRationaleDB.rationale_hash == normalized)
.one_or_none()
)
if row is None:
raise HTTPException(
status_code=404, detail="no rationale recorded for this hash"
)
return RationaleResponse(
rationale_hash=row.rationale_hash,
rationale=row.rationale,
agent_address=row.agent_address,
evaluator_address=row.evaluator_address,
)
|