latest.ipv8.attestation.wallet.community

Module Contents

Classes

AttestationSettings

Settings for the Attestation community.

AttestationCommunity

Community for sharing Attestations.

Functions

synchronized(→ WF)

Due to database inconsistencies, we can't allow multiple threads to handle a received_half_block at the same time.

_default_attestation_request_callback(...)

_default_attestation_request_complete_callback(→ None)

_default_verify_callback(→ asyncio.Future[bool])

Attributes

WF

latest.ipv8.attestation.wallet.community.WF
latest.ipv8.attestation.wallet.community.synchronized(f: WF) → WF

Due to database inconsistencies, we can’t allow multiple threads to handle a received_half_block at the same time.

latest.ipv8.attestation.wallet.community._default_attestation_request_callback(peer: latest.ipv8.types.Peer, attribute_name: str, metadata: dict[str, str]) → asyncio.Future[bytes | None]
latest.ipv8.attestation.wallet.community._default_attestation_request_complete_callback(for_peer: latest.ipv8.types.Peer, attribute_name: str, attr_hash: bytes, id_format: str, from_peer: latest.ipv8.types.Peer | None = None) → None
latest.ipv8.attestation.wallet.community._default_verify_callback(peer: latest.ipv8.types.Peer, attr_hash: bytes) → asyncio.Future[bool]
class latest.ipv8.attestation.wallet.community.AttestationSettings

Bases: latest.ipv8.community.CommunitySettings

Settings for the Attestation community.

working_directory: str = ''
db_name: str = 'attestations'
class latest.ipv8.attestation.wallet.community.AttestationCommunity(settings: AttestationSettings)

Bases: latest.ipv8.community.Community

Community for sharing Attestations.

Note that the logic for giving out Attestations is in the identity chain.

community_id
settings_class
async unload() → None

Shutdown our request cache and database.

get_id_algorithm(id_format: str) → latest.ipv8.types.IdentityAlgorithm

Resolve an algorithm from a name.

set_attestation_request_callback(f: Callable[[latest.ipv8.types.Peer, str, dict[str, str]], asyncio.Future[bytes | None]]) → None

Set the callback to be called when someone requests an attestation from us.

f should accept a (Peer, attribute name, metadata) and return a str()-able value. If it f returns None, no attestation is made.

Parameters:

f – the callback function providing the value

set_attestation_request_complete_callback(f: Callable[[latest.ipv8.types.Peer, str, bytes, str, latest.ipv8.types.Peer | None], None]) → None

f should accept a (Peer, attribute_name, hash, id_format, Peer=None), it is called when an Attestation has been made for another peer.

Parameters:

f – the function to call when an Attestation has been completed

set_verify_request_callback(f: Callable[[latest.ipv8.types.Peer, bytes], asyncio.Future[bool]]) → None

Set the callback to be called when someone wants to verify our attribute.

f should accept a (Peer, attribute_name) and return a boolean value. If f return True, the attribute will be verified.

Parameters:

f – the function to call when our attribute is requested for verification

dump_blob(attribute_name: str, id_format: str, blob: bytes, metadata: dict[str, str] | None = None) → None

Add an attribute directly (without the help of an IPv8 peer).

This is only for advanced use, where the blob already has (1) some form of attestation embedded and (2) follows some form of non-interactive Zero-Knowledge Proof.

Parameters:

  • attribute_name – the attribute we are creating

  • id_format – the identity format

  • blob – the raw data to be processed by the given id_format

  • metadata – optional additional metadata

request_attestation(peer: latest.ipv8.types.Peer, attribute_name: str, secret_key: latest.ipv8.attestation.wallet.database.SecretKeyProtocol, metadata: dict[str, str] | None = None) → None

Request attestation of one of our attributes.

Parameters:

  • peer – Peer of the Attestor

  • attribute_name – the attribute we want attested

  • secret_key – the secret key we use for this attribute

async on_request_attestation(peer: latest.ipv8.types.Peer, dist: latest.ipv8.messaging.payload_headers.GlobalTimeDistributionPayload, payload: latest.ipv8.attestation.wallet.payload.RequestAttestationPayload) → None

Someone wants us to attest their attribute.

on_attestation_complete(unserialized: latest.ipv8.attestation.identity_formats.Attestation, secret_key: latest.ipv8.attestation.wallet.database.SecretKeyProtocol, peer: latest.ipv8.types.Peer, name: str, attestation_hash: bytes, id_format: str) → None

We got an Attestation delivered to us.

verify_attestation_values(socket_address: latest.ipv8.types.Address, attestation_hash: bytes, values: list[bytes], callback: Callable[[bytes, list[float]], None], id_format: str) → None

Ask the peer behind a socket address to deliver the Attestation with a certain hash.

Parameters:

  • socket_address – the socket address to send to

  • attestation_hash – the hash of the Attestation to request

  • values – the values for which we want to measure certainty

  • callback – the callback to call with the map of (hash, {value: certainty})

  • id_format – the identity format specifier

create_verify_attestation_request(socket_address: latest.ipv8.types.Address, attestation_hash: bytes, id_format: str) → None

Ask the peer behind a socket address to deliver the Attestation with a certain hash.

Parameters:

  • socket_address – the socket address to send to

  • attestation_hash – the hash of the Attestation to request

  • id_format – the identity format specifier

async on_verify_attestation_request(peer: latest.ipv8.types.Peer, dist: latest.ipv8.messaging.payload_headers.GlobalTimeDistributionPayload, payload: latest.ipv8.attestation.wallet.payload.VerifyAttestationRequestPayload) → None

We received a request to verify one of our attestations. Send the requested attestation back.

send_attestation(socket_address: latest.ipv8.types.Address, blob: bytes, global_time: int | None = None) → None

Send a serialized attestation (blob) to an address, split into chunks.

If we want to serve this request, send the attestation in chunks of 800 bytes.

on_attestation_chunk(peer: latest.ipv8.types.Peer, dist: latest.ipv8.messaging.payload_headers.GlobalTimeDistributionPayload, payload: latest.ipv8.attestation.wallet.payload.AttestationChunkPayload) → None

We received a chunk of an Attestation.

on_received_attestation(peer: latest.ipv8.types.Peer, attestation: latest.ipv8.attestation.identity_formats.Attestation, attestation_hash: bytes) → None

Callback for when we got the entire attestation from a peer.

Parameters:

  • peer – the Peer we got this attestation from

  • attestation – the Attestation object we can check

on_challenge(peer: latest.ipv8.types.Peer, dist: latest.ipv8.messaging.payload_headers.GlobalTimeDistributionPayload, payload: latest.ipv8.attestation.wallet.payload.ChallengePayload) → None

We received a challenge for an Attestation.

on_challenge_response(peer: latest.ipv8.types.Peer, dist: latest.ipv8.messaging.payload_headers.GlobalTimeDistributionPayload, payload: latest.ipv8.attestation.wallet.payload.ChallengeResponsePayload) → None

We received a response to our challenge.