Deprecated Attestation test scripts

attestation_tutorial_attest.py

import os
import shutil

from attestation_tutorial_common import finish, http_get, http_post, start, urlstr, wait_for_list

# Remove the output of previous experiments.
if os.path.exists('./state_1'):
    shutil.rmtree('./state_1')
if os.path.exists('./state_2'):
    shutil.rmtree('./state_2')

start()
print("Enrollment/Attestation flow")

print("0. SANITY CHECK")
peer1_neighborhood = wait_for_list("http://localhost:14411/attestation?type=peers")
peer2_neighborhood = wait_for_list("http://localhost:14412/attestation?type=peers")

peer1_id = urlstr(peer2_neighborhood[0])
peer2_id = urlstr(peer1_neighborhood[0])

print("Peer 1:", peer1_id)
print("Peer 2:", peer2_id)

print("Peer 1 attributes:", http_get("http://localhost:14411/attestation?type=attributes"))
print("Peer 2 attributes:", http_get("http://localhost:14412/attestation?type=attributes"))

print("1. ATTESTATION REQUEST")
print("Request attestation from peer 2:",
      http_post(f"http://localhost:14411/attestation?type=request&mid={peer2_id}&attribute_name=my_attribute"))

print("2. ATTESTATION")
peer2_outstanding_requests = wait_for_list("http://localhost:14412/attestation?type=outstanding")
print("Peer 2 outstanding requests:", peer2_outstanding_requests)

print("Peer 2 attesting to outstanding request:",
      http_post(f"http://localhost:14412/attestation?type=attest&mid={peer1_id}"
                f"&attribute_name={peer2_outstanding_requests[0][1]}"
                f"&attribute_value=dmFsdWU%3D"))

print("3. CHECK")
print("Peer 1 attributes:", http_get("http://localhost:14411/attestation?type=attributes"))
print("Peer 2 attributes:", http_get("http://localhost:14412/attestation?type=attributes"))

print("X. DONE!")
finish()

attestation_tutorial_common.py

import json
import os
import signal
import subprocess
import time
import urllib.parse
import urllib.request

PROCESS = None


def http_get(url):
    """
    Perform an HTTP GET request to the given URL.
    """
    return json.loads(urllib.request.urlopen(urllib.request.Request(url)).read().decode())


def http_post(url):
    """
    Perform an HTTP POST request to the given URL.
    """
    return json.loads(urllib.request.urlopen(urllib.request.Request(url, method="POST")).read().decode())


def urlstr(s):
    """
    Make the given string URL safe.
    """
    return urllib.parse.quote(s, safe='')


def wait_for_list(url):
    """
    Poll an endpoint until output (a list) is available.
    """
    out = []
    while not out:
        out = http_get(url)
        time.sleep(0.5)
    return out


def start():
    """
    Run the main.py script and wait for it to finish initializing.
    """
    global PROCESS
    PROCESS = subprocess.Popen('python3 main.py', stdout=subprocess.PIPE, shell=True, preexec_fn=os.setsid)
    output = b""
    while output.decode().count("Starting peer") != 2:
        time.sleep(1.0)
        output = PROCESS.stdout.peek()


def finish():
    """
    Kill our two IPv8 instances (running in the same process).
    """
    global PROCESS
    os.killpg(os.getpgid(PROCESS.pid), signal.SIGTERM)
    PROCESS.communicate()

attestation_tutorial_verify.py

import time

from attestation_tutorial_common import finish, http_get, http_post, start, urlstr, wait_for_list

start()
print("Attribute verification flow")

print("0. SANITY CHECK")
peer1_neighborhood = wait_for_list("http://localhost:14411/attestation?type=peers")
peer2_neighborhood = wait_for_list("http://localhost:14412/attestation?type=peers")

peer1_id = urlstr(peer2_neighborhood[0])
peer2_id = urlstr(peer1_neighborhood[0])

print("Peer 1:", peer1_id)
print("Peer 2:", peer2_id)

peer1_attributes = http_get("http://localhost:14411/attestation?type=attributes")
peer2_attributes = http_get("http://localhost:14412/attestation?type=attributes")

print("Peer 1 attributes:", peer1_attributes)
print("Peer 2 attributes:", peer2_attributes)

print("1. VERIFICATION REQUEST")
print("Request verification from peer 1:",
      http_post(f"http://localhost:14412/attestation?type=verify&mid={peer1_id}"
                f"&attribute_hash={urlstr(peer1_attributes[-1][1])}&attribute_values=dmFsdWU%3D"))

print("2. VERIFICATION ")
peer1_outstanding_requests = wait_for_list("http://localhost:14411/attestation?type=outstanding_verify")
print("Peer 1 outstanding verification requests:", peer1_outstanding_requests)

print("Peer 1 allow verification of outstanding request:",
      http_post(f"http://localhost:14411/attestation?type=allow_verify&mid={peer2_id}"
                f"&attribute_name={urlstr(peer1_attributes[-1][0])}"))

print("3. CHECK")
match = 0.0
while match < 0.9:
    for hash, output in http_get("http://localhost:14412/attestation?type=verification_output").items():
        if hash == peer1_attributes[-1][1]:
            match_value, match = output[0]
            assert match_value == "dmFsdWU="
    time.sleep(0.1)
print("Peer 2 verification output:", http_get("http://localhost:14412/attestation?type=verification_output"))

print("X. DONE!")
finish()

main.py

from base64 import b64encode

from asyncio import ensure_future, get_event_loop

from pyipv8.ipv8.configuration import get_default_configuration
from pyipv8.ipv8.REST.rest_manager import RESTManager
from pyipv8.ipv8_service import IPv8


async def start_communities():
    # Launch two IPv8 services.
    # We run REST endpoints for these services on:
    #  - http://localhost:14411/
    #  - http://localhost:14412/
    # This script also prints the peer ids for reference with:
    #  - http://localhost:1441*/attestation?type=peers
    for i in [1, 2]:
        configuration = get_default_configuration()
        configuration['logger']['level'] = "ERROR"
        configuration['keys'] = [
            {'alias': "anonymous id", 'generation': u"curve25519", 'file': u"ec%d_multichain.pem" % i}
        ]

        # Only load the basic communities
        requested_overlays = ['DiscoveryCommunity', 'AttestationCommunity', 'IdentityCommunity']
        configuration['overlays'] = [{
            'class': 'DiscoveryCommunity',
            'key': "anonymous id",
            'walkers': [
                {
                    'strategy': "RandomWalk",
                    'peers': 20,
                    'init': {
                        'timeout': 3.0
                    }
                },
                {
                    'strategy': "RandomChurn",
                    'peers': -1,
                    'init': {
                        'sample_size': 8,
                        'ping_interval': 10.0,
                        'inactive_time': 27.5,
                        'drop_time': 57.5
                    }
                },
                {
                    'strategy': "PeriodicSimilarity",
                    'peers': -1,
                    'init': {}
                }
            ],
            'initialize': {},
            'on_start': [
                ('resolve_dns_bootstrap_addresses',)
            ]
        },
            {
                'class': 'AttestationCommunity',
                'key': "anonymous id",
                'walkers': [{
                    'strategy': "RandomWalk",
                    'peers': 20,
                    'init': {
                        'timeout': 3.0
                    }
                }],
                'initialize': {},
                'on_start': []
            },
            {
                'class': 'IdentityCommunity',
                'key': "anonymous id",
                'walkers': [{
                    'strategy': "RandomWalk",
                    'peers': 20,
                    'init': {
                        'timeout': 3.0
                    }
                }],
                'initialize': {},
                'on_start': []
            }]

        # Give each peer a separate working directory
        working_directory_overlays = ['AttestationCommunity', 'IdentityCommunity']
        for overlay in configuration['overlays']:
            if overlay['class'] in working_directory_overlays:
                overlay['initialize'] = {'working_directory': 'state_%d' % i}

        # Start the IPv8 service
        ipv8 = IPv8(configuration)
        await ipv8.start()
        rest_manager = RESTManager(ipv8)
        await rest_manager.start(14410 + i)

        # Print the peer for reference
        print("Starting peer", b64encode(ipv8.keys["anonymous id"].mid))


ensure_future(start_communities())
get_event_loop().run_forever()