Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
92 changes: 91 additions & 1 deletion src/keria/app/credentialing.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,9 @@ def loadEnds(app, identifierResource):
credentialVerificationEnd = CredentialVerificationCollectionEnd()
app.add_route("/credentials/verify", credentialVerificationEnd)

verificationEnd = VerificationCollectionEnd()
app.add_route("/verify", verificationEnd)


class EmptyDictSchema(MarshmallowSchema):
class Meta:
Expand Down Expand Up @@ -704,8 +707,9 @@ def on_post(req, rep):

---
summary: Verify a credential without IPEX
description: Verify a credential without using IPEX (TEL should be updated separately)
description: Deprecated - use the generic POST /verify endpoint instead. Verify a credential without using IPEX (TEL should be updated separately)
operationId: verifyCredential
deprecated: true
tags:
- Credentials
requestBody:
Expand Down Expand Up @@ -735,6 +739,10 @@ def on_post(req, rep):
404:
description: Malformed ACDC or iss event
"""
# RFC 9745: deprecated in favor of the generic POST /verify endpoint
rep.set_header("Deprecation", "@1784073600") # 2026-07-15T00:00:00Z
rep.set_header("Link", '</verify>; rel="successor-version"')

agent = req.context.agent
body = req.get_media()

Expand All @@ -758,6 +766,88 @@ def on_post(req, rep):
rep.data = op.to_json().encode("utf-8")


class VerificationCollectionEnd:
"""Generic verification endpoint.

Accepts an ACDC or TEL event (as a KED) and its CESR attachments, returns a
long running operation whose type is selected by the Serder's ilk.
"""

@staticmethod
def on_post(req, rep):
"""Verify a Serder

---
summary: Verify a Serder (credential, registry, ...) by its ilk
description:
Accepts an ACDC or TEL event (as a KED) and its CESR attachments,
returns a long running operation dispatched by the Serder's ilk.
operationId: verify
tags:
- Credentials
requestBody:
required: true
content:
application/json:
schema:
type: object
required:
- serder
- atc
properties:
serder:
type: object
description: KED of ACDC or TEL event
atc:
type: string
description: CESR attachments for the Serder
responses:
202:
description: Serder accepted for verification; long running operation returned
content:
application/json:
schema:
description: long running operation of the verification
400:
description: Malformed Serder or unsupported ilk
"""
agent = req.context.agent
body = req.get_media()

try:
serder = serdering.Serder(sad=httping.getRequiredParam(body, "serder"))
except (kering.KeriError, TypeError) as e:
rep.status = falcon.HTTP_400
rep.text = e.args[0] if e.args else str(e)
return

atc = httping.getRequiredParam(body, "atc")

if serder.proto == Protocols.acdc:
oid = serder.said
optype = longrunning.OpTypes.credential
metadata = dict(ced=serder.sad)
elif serder.ilk == coring.Ilks.vcp:
regk = serder.sad["i"]
oid = regk
optype = longrunning.OpTypes.registry
metadata = dict(pre=serder.sad["ii"], anchor=dict(i=regk, s="0", d=regk))
elif serder.ilk == coring.Ilks.iss:
vcid = serder.sad["i"]
oid = vcid
optype = longrunning.OpTypes.credential
metadata = dict(ced=dict(d=vcid))
else:
raise falcon.HTTPBadRequest(
description=f"unsupported ilk '{serder.ilk or serder.proto}' for verification"
)

agent.parser.ims.extend(serder.raw + atc.encode("utf-8"))
op = agent.monitor.submit(oid, optype, metadata=metadata)
rep.status = falcon.HTTP_202
rep.data = op.to_json().encode("utf-8")


class CredentialQueryCollectionEnd:
"""This class provides a collection endpoint for creating credential queries.

Expand Down
2 changes: 1 addition & 1 deletion src/keria/core/longrunning.py
Original file line number Diff line number Diff line change
Expand Up @@ -477,7 +477,7 @@ def status(self, op):
)

ced = op.metadata["ced"]
if self.credentialer.complete(ced["d"]):
if self.credentialer.rgy.reger.saved.get(keys=ced["d"]) is not None:

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

credentialer.complete only works for credentials we issue - looking at the escrows, there isn't really much difference right now compared to reger.saved which would allow this op type to be used for general verification of credentials too.

Not sure if this would change when we have deployed registrars is the only thing. Happy to split to two if we think that'd be better.

done = True
response = dict(ced=ced)
else:
Expand Down
89 changes: 88 additions & 1 deletion tests/app/test_credentialing.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,14 @@
import json

import falcon
import pytest
from falcon import testing
from hio.base import doing
from keri.app import habbing
from keri.core import scheming, coring, parsing, serdering
from keri.core.eventing import SealEvent
from keri.core.signing import Salter
from keri.kering import TraitCodex
from keri.kering import KeriError, TraitCodex
from keri.vc import proving
from keri.vdr import eventing
from keri.vdr.credentialing import Regery, Registrar
Expand All @@ -38,6 +39,92 @@ def test_load_ends(helpers):
assert isinstance(end, credentialing.SchemaResourceEnd)
(end, *_) = app._router.find("/identifiers/NAME/registries")
assert isinstance(end, credentialing.RegistryCollectionEnd)
(end, *_) = app._router.find("/credentials/verify")
assert isinstance(end, credentialing.CredentialVerificationCollectionEnd)
(end, *_) = app._router.find("/verify")
assert isinstance(end, credentialing.VerificationCollectionEnd)


def test_verify_end(helpers):
salt = b"0123456789abcdef"
with helpers.openKeria() as (agency, agent, app, client):
app.add_route("/identifiers", aiding.IdentifierCollectionEnd())
app.add_route("/verify", credentialing.VerificationCollectionEnd())

op = helpers.createAid(client, "issuer", salt)
issuerPre = op["response"]["i"]
assert issuerPre in agent.hby.kevers

# registry inception (vcp) -> registry operation
vcp = eventing.incept(issuerPre)
body = dict(serder=vcp.sad, atc="")
res = client.simulate_post("/verify", body=json.dumps(body).encode("utf-8"))
assert res.status_code == 202
rop = res.json
assert rop["name"] == f"registry.{vcp.pre}"
assert rop["metadata"]["pre"] == issuerPre
assert rop["metadata"]["anchor"] == dict(i=vcp.pre, s="0", d=vcp.pre)

# TEL issuance (iss) -> credential operation keyed on the credential SAID
iss = eventing.issue(
vcdig="EBfdlu8R27Fbx-ehrqwImnK-8Cm79sqbAQ4MmvEAYqao", regk=vcp.pre
)
body = dict(serder=iss.sad, atc="")
res = client.simulate_post("/verify", body=json.dumps(body).encode("utf-8"))
assert res.status_code == 202
iop = res.json
assert iop["name"] == f"credential.{iss.sad['i']}"
assert iop["metadata"]["ced"]["d"] == iss.sad["i"]

# ACDC -> credential operation
creder = serdering.SerderACDC(
sad=dict(
v="ACDC10JSON000000_",
d="",
i=issuerPre,
ri=vcp.pre,
s="EBfdlu8R27Fbx-ehrqwImnK-8Cm79sqbAQ4MmvEAYqao",
a={},
),
makify=True,
)
body = dict(serder=creder.sad, atc="")
res = client.simulate_post("/verify", body=json.dumps(body).encode("utf-8"))
assert res.status_code == 202
cop = res.json
assert cop["name"] == f"credential.{creder.said}"
assert cop["metadata"]["ced"] == creder.sad

# unsupported ilk (icp) -> 400
icp = agent.hby.kevers[issuerPre].serder
body = dict(serder=icp.sad, atc="")
res = client.simulate_post("/verify", body=json.dumps(body).encode("utf-8"))
assert res.status_code == 400

# 'serder' that is not a valid Serder field map -> 400
# TypeError: valid JSON but not an object (number, string, array)
# KeriError: empty sad, missing/invalid "v" version string, bad SAID
tampered = dict(vcp.sad, d="E" + "A" * 43)
for value, etype in (
(123, TypeError),
("v str", TypeError),
(["v"], TypeError),
({}, KeriError),
({"a": 1}, KeriError),
({"v": "bogus"}, KeriError),
(tampered, KeriError),
):
with pytest.raises(etype):
serdering.Serder(sad=value)
body = dict(serder=value, atc="")
res = client.simulate_post("/verify", body=json.dumps(body).encode("utf-8"))
assert res.status_code == 400

# missing required param -> 400
res = client.simulate_post(
"/verify", body=json.dumps(dict(atc="")).encode("utf-8")
)
assert res.status_code == 400


def test_schema_ends(helpers):
Expand Down
Loading