3.0.ipv8.attestation.wallet.community ===================================== .. py:module:: 3.0.ipv8.attestation.wallet.community Attributes ---------- .. autoapisummary:: 3.0.ipv8.attestation.wallet.community.WF Classes ------- .. autoapisummary:: 3.0.ipv8.attestation.wallet.community.AttestationSettings 3.0.ipv8.attestation.wallet.community.AttestationCommunity Functions --------- .. autoapisummary:: 3.0.ipv8.attestation.wallet.community.synchronized 3.0.ipv8.attestation.wallet.community._default_attestation_request_callback 3.0.ipv8.attestation.wallet.community._default_attestation_request_complete_callback 3.0.ipv8.attestation.wallet.community._default_verify_callback Module Contents --------------- .. py:data:: WF .. py:function:: synchronized(f: WF) -> WF Due to database inconsistencies, we can't allow multiple threads to handle a received_half_block at the same time. .. py:function:: _default_attestation_request_callback(peer: 3.0.ipv8.types.Peer, attribute_name: str, metadata: dict[str, str]) -> asyncio.Future[bytes | None] .. py:function:: _default_attestation_request_complete_callback(for_peer: 3.0.ipv8.types.Peer, attribute_name: str, attr_hash: bytes, id_format: str, from_peer: 3.0.ipv8.types.Peer | None = None) -> None .. py:function:: _default_verify_callback(peer: 3.0.ipv8.types.Peer, attr_hash: bytes) -> asyncio.Future[bool] .. py:class:: AttestationSettings Bases: :py:obj:`3.0.ipv8.community.CommunitySettings` Settings for the Attestation community. .. py:attribute:: working_directory :type: str :value: '' .. py:attribute:: db_name :type: str :value: 'attestations' .. py:class:: AttestationCommunity(settings: AttestationSettings) Bases: :py:obj:`3.0.ipv8.community.Community` Community for sharing Attestations. Note that the logic for giving out Attestations is in the identity chain. .. py:attribute:: community_id .. py:attribute:: settings_class .. py:attribute:: receive_block_lock .. py:attribute:: schema_manager .. py:attribute:: attestation_request_callback :type: Callable[[3.0.ipv8.types.Peer, str, dict[str, str]], asyncio.Future[bytes | None]] .. py:attribute:: attestation_request_complete_callback :type: Callable[[3.0.ipv8.types.Peer, str, bytes, str, 3.0.ipv8.types.Peer | None], None] .. py:attribute:: verify_request_callback :type: Callable[[3.0.ipv8.types.Peer, bytes], asyncio.Future[bool]] .. py:attribute:: attestation_keys :type: dict[bytes, tuple[3.0.ipv8.attestation.wallet.database.SecretKeyProtocol, str]] .. py:attribute:: database .. py:attribute:: cached_attestation_blobs :type: dict[bytes, 3.0.ipv8.attestation.identity_formats.Attestation] .. py:attribute:: allowed_attestations :type: dict[bytes, list[bytes]] .. py:attribute:: request_cache .. py:method:: unload() -> None :async: Shutdown our request cache and database. .. py:method:: get_id_algorithm(id_format: str) -> 3.0.ipv8.types.IdentityAlgorithm Resolve an algorithm from a name. .. py:method:: set_attestation_request_callback(f: Callable[[3.0.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. :param f: the callback function providing the value .. py:method:: set_attestation_request_complete_callback(f: Callable[[3.0.ipv8.types.Peer, str, bytes, str, 3.0.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. :param f: the function to call when an Attestation has been completed .. py:method:: set_verify_request_callback(f: Callable[[3.0.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. :param f: the function to call when our attribute is requested for verification .. py:method:: 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. :param attribute_name: the attribute we are creating :param id_format: the identity format :param blob: the raw data to be processed by the given id_format :param metadata: optional additional metadata .. py:method:: request_attestation(peer: 3.0.ipv8.types.Peer, attribute_name: str, secret_key: 3.0.ipv8.attestation.wallet.database.SecretKeyProtocol, metadata: dict[str, str] | None = None) -> None Request attestation of one of our attributes. :param peer: Peer of the Attestor :param attribute_name: the attribute we want attested :param secret_key: the secret key we use for this attribute .. py:method:: on_request_attestation(peer: 3.0.ipv8.types.Peer, dist: 3.0.ipv8.messaging.payload_headers.GlobalTimeDistributionPayload, payload: 3.0.ipv8.attestation.wallet.payload.RequestAttestationPayload) -> None :async: Someone wants us to attest their attribute. .. py:method:: on_attestation_complete(unserialized: 3.0.ipv8.attestation.identity_formats.Attestation, secret_key: 3.0.ipv8.attestation.wallet.database.SecretKeyProtocol, peer: 3.0.ipv8.types.Peer, name: str, attestation_hash: bytes, id_format: str) -> None We got an Attestation delivered to us. .. py:method:: verify_attestation_values(socket_address: 3.0.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. :param socket_address: the socket address to send to :param attestation_hash: the hash of the Attestation to request :param values: the values for which we want to measure certainty :param callback: the callback to call with the map of (hash, {value: certainty}) :param id_format: the identity format specifier .. py:method:: create_verify_attestation_request(socket_address: 3.0.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. :param socket_address: the socket address to send to :param attestation_hash: the hash of the Attestation to request :param id_format: the identity format specifier .. py:method:: on_verify_attestation_request(peer: 3.0.ipv8.types.Peer, dist: 3.0.ipv8.messaging.payload_headers.GlobalTimeDistributionPayload, payload: 3.0.ipv8.attestation.wallet.payload.VerifyAttestationRequestPayload) -> None :async: We received a request to verify one of our attestations. Send the requested attestation back. .. py:method:: send_attestation(socket_address: 3.0.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. .. py:method:: on_attestation_chunk(peer: 3.0.ipv8.types.Peer, dist: 3.0.ipv8.messaging.payload_headers.GlobalTimeDistributionPayload, payload: 3.0.ipv8.attestation.wallet.payload.AttestationChunkPayload) -> None We received a chunk of an Attestation. .. py:method:: on_received_attestation(peer: 3.0.ipv8.types.Peer, attestation: 3.0.ipv8.attestation.identity_formats.Attestation, attestation_hash: bytes) -> None Callback for when we got the entire attestation from a peer. :param peer: the Peer we got this attestation from :param attestation: the Attestation object we can check .. py:method:: on_challenge(peer: 3.0.ipv8.types.Peer, dist: 3.0.ipv8.messaging.payload_headers.GlobalTimeDistributionPayload, payload: 3.0.ipv8.attestation.wallet.payload.ChallengePayload) -> None We received a challenge for an Attestation. .. py:method:: on_challenge_response(peer: 3.0.ipv8.types.Peer, dist: 3.0.ipv8.messaging.payload_headers.GlobalTimeDistributionPayload, payload: 3.0.ipv8.attestation.wallet.payload.ChallengeResponsePayload) -> None We received a response to our challenge.