Loading vLEI.wiki
Fetching knowledge base...
Fetching knowledge base...
This comprehensive explanation has been generated from 3 GitHub source documents. All source documents are searchable here.
Last updated: September 21, 2025
This content is meant to be consumed by AI agents via MCP. Click here to get the MCP configuration.
Note: In rare cases it may contain LLM hallucinations.
For authoritative documentation, please consult the official GLEIF vLEI trainings and the ToIP Glossary.
A standardized media type identifier that defines the foundational format for verifiable credentials, establishing credential+ld+json as the canonical base type with mandatory transformation requirements for alternative formats like credential+acdc+json.
The base-media-type is a foundational specification within the KERI/ACDC ecosystem that establishes credential+ld+json as the canonical media type for verifiable credentials. This base type serves as the reference format against which all other credential media types must provide either unidirectional or bidirectional transformations, ensuring semantic interoperability across the decentralized identity ecosystem.
base_media_type := "credential" "+" "ld" "+" "json"
alternative_type := "credential" "+" format_specifier "+" encoding
transformation_requirement := unidirectional_transform | bidirectional_transform
The specification mandates that any alternative media type T_alt must satisfy:
∃ f: T_alt → credential+ld+json (unidirectional transformation)∃ f: T_alt ↔ credential+ld+json (bidirectional transformation)The base media type follows a three-component composition pattern:
credential - Specifies the semantic domainld - Indicates Linked Data format complianceProblem: Malicious @context URLs can redefine credential semantics Solution: Implement context allowlisting and semantic validation
ALLOWED_CONTEXTS = {
'https://www.w3.org/2018/credentials/v1',
'https://www.gleif.org/vlei/v1',
'https://schema.org/'
}
def validate_contexts(contexts: List[str]) -> bool:
return all(ctx in ALLOWED_CONTEXTS for ctx in contexts)
Problem: Concurrent transformations can corrupt shared state Solution: Use immutable data structures and thread-local storage
import threading
from copy import deepcopy
local_data = threading.local()
def safe_transform(credential: dict) -> dict:
local_data.working_copy = deepcopy(credential)
return transform_internal(local_data.working_copy)
Problem: Large credentials can cause OOM during transformation Solution: Implement streaming transformation with size limits
MAX_CREDENTIAL_SIZE = 1024 * 1024 # 1MB
def validate_size(credential_data: bytes) -> None:
if len(credential_data) > MAX_CREDENTIAL_SIZE:
raise ValueError(f"Credential exceeds size limit: {len(credential_data)}")
from functools import lru_cache
import asyncio
@lru_cache(maxsize=1000)
def get_cached_schema(schema_url: str) -> dict:
return fetch_schema(schema_url)
# Async preloading for hot schemas
async def preload_schemas():
hot_schemas = get_frequently_used_schemas()
tasks = [fetch_schema(url) for url in hot_schemas]
await asyncio.gather(*tasks)
def batch_transform(credentials: List[dict],
source_type: str,
target_type: str) -> List[dict]:
handler = get_handler(source_type)
# Pre-validate all credentials
invalid_indices = []
for i, cred in enumerate(credentials):
if not handler.validate(cred):
invalid_indices.append(i)
if invalid_indices:
raise ValidationError(f"Invalid credentials at indices: {invalid_indices}")
# Batch transform with parallel processing
with ThreadPoolExecutor(max_workers=4) as executor:
results = list(executor.map(handler.transform_to_base, credentials))
return results
jsonThe KERI ecosystem defines credential+acdc+json as the preferred alternative format with specific architectural decisions:
acdc_media_type := "credential" "+" "acdc" "+" "json"
context_exclusion := true // Deliberate @context omission
transformation_type := unidirectional // To credential+ld+json
{
"@context": [
"https://www.w3.org/2018/credentials/v1",
"<additional-context-urls>"
],
"type": ["VerifiableCredential", "<specific-credential-type>"],
"issuer": "<issuer-identifier>",
"issuanceDate": "<ISO-8601-datetime>",
"credentialSubject": {
"id": "<subject-identifier>",
"<claim-properties>": "<claim-values>"
},
"proof": {
"type": "<proof-type>",
"created": "<ISO-8601-datetime>",
"proofPurpose": "<proof-purpose>",
"verificationMethod": "<verification-method-id>",
"<proof-specific-fields>": "<proof-values>"
}
}
{
"v": "ACDC10JSON000000_",
"d": "<SAID-identifier>",
"i": "<issuer-AID>",
"ri": "<registry-identifier>",
"s": "<schema-SAID>",
"a": {
"<attribute-name>": "<attribute-value>",
"<nested-attributes>": {
"<sub-attribute>": "<sub-value>"
}
},
"e": {
"<edge-name>": {
"n": "<target-SAID>",
"s": "<edge-schema-SAID>"
}
},
"r": {
"<rule-name>": "<rule-value>"
}
}
sequenceDiagram
participant Client
participant Transformer
participant Validator
Client->>Transformer: credential+acdc+json
Transformer->>Transformer: Parse ACDC structure
Transformer->>Transformer: Map to LD-JSON schema
Transformer->>Transformer: Generate @context
Transformer->>Validator: credential+ld+json
Validator->>Validator: Validate W3C compliance
Validator->>Client: Validation result
The protocol supports content negotiation through HTTP Accept headers:
Accept: application/credential+acdc+json;q=0.9, application/credential+ld+json;q=0.8
Content-Type: application/credential+acdc+json
States: {DETECT, VALIDATE, TRANSFORM, PROCESS, ERROR}
DETECT:
input: media_type_string
if media_type == "credential+ld+json" → VALIDATE
if media_type == "credential+acdc+json" → TRANSFORM
else → ERROR
VALIDATE:
input: credential_data
if valid_w3c_format → PROCESS
else → ERROR
TRANSFORM:
input: acdc_credential
output: ld_json_credential
→ VALIDATE
PROCESS:
input: validated_credential
output: processed_result
ERROR:
output: error_response
ACDC credentials use Self-Addressing Identifiers (SAIDs) for cryptographic integrity:
def generate_said(content: dict) -> str:
"""
Generate SAID for ACDC content
"""
# Remove existing 'd' field for calculation
temp_content = content.copy()
temp_content.pop('d', None)
# Serialize with canonical ordering
canonical_json = json.dumps(temp_content,
separators=(',', ':'),
sort_keys=True)
# Calculate Blake3-256 hash
hash_bytes = blake3(canonical_json.encode('utf-8')).digest()[:32]
# Encode as CESR primitive
return encode_cesr_primitive('E', hash_bytes) # 'E' = Blake3-256
def verify_acdc_signature(acdc: dict, signature: str, issuer_aid: str) -> bool:
"""
Verify ACDC signature against issuer's current key state
"""
# 1. Resolve issuer's current key state from KEL
key_state = resolve_key_state(issuer_aid)
# 2. Extract verification key
verification_key = key_state.current_keys[0] # Simplified
# 3. Reconstruct signing data
signing_data = construct_signing_data(acdc)
# 4. Verify signature
return ed25519_verify(verification_key, signing_data, signature)
POST /api/v1/media-types/register
Content-Type: application/json
{
"mediaType": "credential+custom+json",
"transformationEndpoint": "https://transformer.example.com/to-ld-json",
"transformationType": "unidirectional",
"schemaUrl": "https://schema.example.com/custom-credential.json"
}
POST /api/v1/transform
Content-Type: application/credential+acdc+json
Accept: application/credential+ld+json
{
"v": "ACDC10JSON000000_",
"d": "EBdXt3gIXOf2BBWNHdSXCJnFJL5OuQPyM5K0neuniccM",
"i": "EaU6JR2nmwyZ-i0d8JZAoTNZH3ULvYAfSVPzhzS6b5CM",
"s": "E46jrVPTzlSkUPqGGeIZ8a8FWS7a6s4reAXRZOkogZ2A",
"a": {
"LEI": "5493001KJTIIGC8Y1R17"
}
}
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
class MediaTypeHandler(ABC):
"""
Abstract base class for media type handlers
"""
@abstractmethod
def get_media_type(self) -> str:
"""Return the media type string this handler supports"""
pass
@abstractmethod
def validate(self, content: Dict[str, Any]) -> bool:
"""Validate content against media type schema"""
pass
@abstractmethod
def transform_to_base(self, content: Dict[str, Any]) -> Dict[str, Any]:
"""Transform content to base media type format"""
pass
@abstractmethod
def transform_from_base(self, content: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Transform from base type (None if unidirectional)"""
pass
class ACDCHandler(MediaTypeHandler):
"""
Handler for credential+acdc+json media type
"""
def get_media_type(self) -> str:
return "credential+acdc+json"
def validate(self, content: Dict[str, Any]) -> bool:
required_fields = ['v', 'd', 'i', 's']
return all(field in content for field in required_fields)
def transform_to_base(self, content: Dict[str, Any]) -> Dict[str, Any]:
"""
Transform ACDC to W3C VC format
"""
return {
"@context": [
"https://www.w3.org/2018/credentials/v1",
"https://www.gleif.org/vlei/v1"
],
"type": ["VerifiableCredential", "LegalEntityCredential"],
"issuer": content['i'],
"issuanceDate": self._extract_issuance_date(content),
"credentialSubject": {
"id": content.get('a', {}).get('i', ''),
**content.get('a', {})
},
"credentialSchema": {
"id": content['s'],
"type": "JsonSchemaValidator2018"
}
}
def transform_from_base(self, content: Dict[str, Any]) -> None:
# Unidirectional transformation only
return None
| Component | Version | Purpose | Critical |
|---|---|---|---|
| KERIpy | ≥1.0.0 | AID resolution, KEL processing | Yes |
| CESRide | ≥0.3.0 | Primitive encoding/decoding | Yes |
| jsonschema | ≥4.0.0 | Schema validation | Yes |
| blake3 | ≥0.3.0 | SAID generation | Yes |
| ed25519 | ≥1.5.0 | Signature verification | Yes |
media_types:
base_type: "credential+ld+json"
supported_alternatives:
- type: "credential+acdc+json"
transformation: "unidirectional"
handler: "ACDCHandler"
context_required: false
- type: "credential+jwt+json"
transformation: "bidirectional"
handler: "JWTHandler"
context_required: true
transformation:
timeout_seconds: 30
max_content_size: 1048576 # 1MB
cache_ttl: 3600 # 1 hour
validation:
strict_mode: true
allow_unknown_fields: false
schema_cache_size: 1000
The base-media-type concept integrates with multiple KERI components:
KERI Identifier (AID) ←→ Credential Issuer Field
↓
Key Event Log (KEL) ←→ Signature Verification
↓
Witness Network ←→ Duplicity Detection
↓
ACDC Schema Registry ←→ Schema Resolution
↓
CESR Encoding ←→ Primitive Serialization
graph TD
A[Credential Request] --> B{Media Type Detection}
B -->|credential+ld+json| C[Direct Processing]
B -->|credential+acdc+json| D[ACDC Handler]
B -->|Unknown Type| E[Error Response]
D --> F[ACDC Validation]
F --> G[Transform to LD-JSON]
G --> H[W3C Validation]
H --> C
C --> I[Signature Verification]
I --> J[KEL Resolution]
J --> K[Final Validation]
K --> L[Response]
# Typical memory usage for credential processing
class MemoryProfile:
credential_base: int = 2048 # 2KB average credential
transformation_overhead: int = 512 # 512B for transformation
validation_cache: int = 16384 # 16KB schema cache
signature_context: int = 256 # 256B crypto context
total_per_credential = (
credential_base +
transformation_overhead +
validation_cache +
signature_context
) # ~19KB per credential
class MediaTypeError(Exception):
"""Raised when media type is malformed or unsupported"""
pass
def parse_media_type(media_type_string: str) -> MediaType:
try:
parts = media_type_string.split('+')
if len(parts) != 3:
raise MediaTypeError(f"Invalid format: {media_type_string}")
domain, format_spec, encoding = parts
if domain != "credential":
raise MediaTypeError(f"Unsupported domain: {domain}")
return MediaType(domain, format_spec, encoding)
except Exception as e:
raise MediaTypeError(f"Parse error: {str(e)}")
class TransformationError(Exception):
"""Raised when credential transformation fails"""
pass
def safe_transform(handler: MediaTypeHandler, content: dict) -> dict:
try:
# Validate input
if not handler.validate(content):
raise TransformationError("Input validation failed")
# Attempt transformation
result = handler.transform_to_base(content)
# Validate output
if not validate_w3c_format(result):
raise TransformationError("Output validation failed")
return result
except Exception as e:
raise TransformationError(f"Transformation failed: {str(e)}")
The ACDC approach deliberately excludes @context to prevent semantic ambiguity:
def detect_context_ambiguity(ld_json_cred: dict) -> List[str]:
"""
Detect potential semantic ambiguities in @context
"""
warnings = []
contexts = ld_json_cred.get('@context', [])
# Check for conflicting term definitions
term_definitions = {}
for context_url in contexts:
context_doc = resolve_context(context_url)
for term, definition in context_doc.items():
if term in term_definitions:
if term_definitions[term] != definition:
warnings.append(f"Conflicting definition for '{term}'")
term_definitions[term] = definition
return warnings
The base media type ensures compliance with:
apiVersion: apps/v1
kind: Deployment
metadata:
name: credential-processor
spec:
replicas: 3
selector:
matchLabels:
app: credential-processor
template:
spec:
containers:
- name: processor
image: keri/credential-processor:v1.2.0
env:
- name: MEDIA_TYPE_CONFIG
value: "/config/media-types.yaml"
- name: TRANSFORMATION_TIMEOUT
value: "30"
resources:
requests:
memory: "256Mi"
cpu: "100m"
limits:
memory: "512Mi"
cpu: "500m"
livenessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 30
periodSeconds: 10
import prometheus_client
from prometheus_client import Counter, Histogram, Gauge
# Metrics collection
CREDENTIAL_TRANSFORMATIONS = Counter(
'credential_transformations_total',
'Total credential transformations',
['source_type', 'target_type', 'status']
)
TRANSFORMATION_DURATION = Histogram(
'transformation_duration_seconds',
'Time spent transforming credentials',
['source_type', 'target_type']
)
ACTIVE_TRANSFORMATIONS = Gauge(
'active_transformations',
'Currently active transformations'
)
def monitor_transformation(source_type: str, target_type: str):
def decorator(func):
def wrapper(*args, **kwargs):
ACTIVE_TRANSFORMATIONS.inc()
with TRANSFORMATION_DURATION.labels(
source_type=source_type,
target_type=target_type
).time():
try:
result = func(*args, **kwargs)
CREDENTIAL_TRANSFORMATIONS.labels(
source_type=source_type,
target_type=target_type,
status='success'
).inc()
return result
except Exception as e:
CREDENTIAL_TRANSFORMATIONS.labels(
source_type=source_type,
target_type=target_type,
status='error'
).inc()
raise
finally:
ACTIVE_TRANSFORMATIONS.dec()
return wrapper
return decorator
import re
from html import escape
def sanitize_credential_field(value: str) -> str:
# Remove potentially dangerous characters
sanitized = re.sub(r'[<>"\'\/]', '', value)
# HTML escape remaining content
return escape(sanitized)
def sanitize_credential(credential: dict) -> dict:
sanitized = {}
for key, value in credential.items():
if isinstance(value, str):
sanitized[key] = sanitize_credential_field(value)
elif isinstance(value, dict):
sanitized[key] = sanitize_credential(value)
else:
sanitized[key] = value
return sanitized
def verify_credential_chain(credential: dict,
issuer_aid: str,
keri_client: KERIClient) -> bool:
# 1. Resolve current key state
key_state = keri_client.get_key_state(issuer_aid)
if not key_state:
return False
# 2. Check for duplicity
if keri_client.check_duplicity(issuer_aid):
return False
# 3. Verify signature against current keys
signing_data = construct_signing_data(credential)
signature = extract_signature(credential)
return any(
verify_signature(key, signing_data, signature)
for key in key_state.current_keys
)
from hypothesis import given, strategies as st
@given(st.dictionaries(
st.text(min_size=1, max_size=50),
st.one_of(st.text(), st.integers(), st.booleans())
))
def test_transformation_preserves_data(credential_data):
# Add required ACDC fields
acdc_cred = {
'v': 'ACDC10JSON000000_',
'd': 'EBdXt3gIXOf2BBWNHdSXCJnFJL5OuQPyM5K0neuniccM',
'i': 'EaU6JR2nmwyZ-i0d8JZAoTNZH3ULvYAfSVPzhzS6b5CM',
's': 'E46jrVPTzlSkUPqGGeIZ8a8FWS7a6s4reAXRZOkogZ2A',
'a': credential_data
}
# Transform to W3C format
w3c_cred = transform_acdc_to_w3c(acdc_cred)
# Verify data preservation
for key, value in credential_data.items():
assert key in w3c_cred['credentialSubject']
assert w3c_cred['credentialSubject'][key] == value
import pytest
from unittest.mock import Mock, patch
class TestMediaTypeIntegration:
@pytest.fixture
def mock_keri_client(self):
client = Mock()
client.get_key_state.return_value = Mock(
current_keys=['EaU6JR2nmwyZ-i0d8JZAoTNZH3ULvYAfSVPzhzS6b5CM']
)
client.check_duplicity.return_value = False
return client
@patch('requests.get')
def test_schema_resolution(self, mock_get, mock_keri_client):
mock_get.return_value.json.return_value = {
'$schema': 'http://json-schema.org/draft-07/schema#',
'type': 'object'
}
processor = CredentialProcessor(mock_keri_client)
result = processor.process_credential(
test_acdc_credential(),
'credential+acdc+json'
)
assert result['status'] == 'valid'
assert 'credentialSubject' in result['credential']