Testing anoncreds changes

anoncreds_test.py

import os
import time

from controller.controller import Controller
from controller.protocols import (
    indy_anoncred_onboard,
    didexchange,
    indy_issue_credential_v2,
    indy_present_proof_v2,
)
from controller.logging import logging_to_stdout

ALICE = os.getenv("ALICE", "http://alice:3001")
BOB = os.getenv("BOB", "http://bob:3005")

async def main():
    logging_to_stdout()
    async with Controller(base_url=ALICE) as alice, Controller(base_url=BOB) as bob:
        # DID Setup
        public_did = await indy_anoncred_onboard(alice)

        # Register a Schema using legacy Indy
        response = await alice.post(
            "/anoncreds/schema",
            json={
                "schema": {
                    "attrNames": ["name", "age"],
                    "issuerId": public_did.did,
                    "name": "anoncreds-testing",
                    "version": "0.1",
                },
                "options": {},
            },
        )
        schema_id = response["schema_state"]["schema_id"]
        schema = await alice.get(f"/anoncreds/schema/{schema_id}")
        schemas = await alice.get("/anoncreds/schemas")

        cred_def = await alice.post(
            "/anoncreds/credential-definition",
            json={
                "credential_definition": {
                    "tag": "default",
                    "schemaId": schema_id,
                    "issuerId": public_did.did,
                },
                "options": {
                    "support_revocation": True,
                    "max_cred_num": 10,
                },
            },
        )
        cred_def_id = cred_def["credential_definition_state"][
            "credential_definition_id"
        ]
        cred_def = await alice.get(f"/anoncreds/credential-definition/{cred_def_id}")
        cred_defs = await alice.get("/anoncreds/credential-definitions")

        alice_conn, bob_conn = await didexchange(alice, bob)
        alice_cred_ex, bob_cred_ex = await indy_issue_credential_v2(
            alice,
            bob,
            alice_conn.connection_id,
            bob_conn.connection_id,
            cred_def_id,
            {"name": "Bob", "age": "42"},
        )
        bob_pres, alice_pres = await indy_present_proof_v2(
            bob,
            alice,
            bob_conn.connection_id,
            alice_conn.connection_id,
            name="proof-1",
            version="0.1",
            comment="testing",
            requested_attributes=[
                {"name": "name", "restrictions": [{"cred_def_id": cred_def_id}]},
                {"name": "age", "restrictions": [{"cred_def_id": cred_def_id}]},
            ],
        )
        print("Before revocation")
        print(alice_pres.verified, "should be true")
        before_revoking_time = int(time.time())

        await asyncio.sleep(5)

        result = await alice.post(
            "/anoncreds/revoke",
            json={
                "cred_ex_id": alice_cred_ex.cred_ex_id,
                "connection_id": alice_conn.connection_id,
                "notify": True,
            },
        )
        result = await alice.post(
            "/anoncreds/publish-revocations",
        )
        await asyncio.sleep(3)

        # Request proof from holder again after revoking
        revoked_time = int(time.time())
        bob_pres, alice_pres = await indy_present_proof_v2(
            bob,
            alice,
            bob_conn.connection_id,
            alice_conn.connection_id,
            requested_attributes=[
                {
                    "name": "name",
                    "restrictions": [{"cred_def_id": cred_def_id}],
                }
            ],
            non_revoked={"from": revoked_time, "to": revoked_time},
        )
        print("Interval after revocation")
        print(alice_pres.verified, "should be false")

        # Request proof from holder again after revoking,
        # using the interval before cred revoked
        # (non_revoked interval/when cred was valid)
        bob_pres, alice_pres = await indy_present_proof_v2(
            bob,
            alice,
            bob_conn.connection_id,
            alice_conn.connection_id,
            requested_attributes=[
                {
                    "name": "name",
                    "restrictions": [{"cred_def_id": cred_def_id}],
                }
            ],
            non_revoked={"from": before_revoking_time, "to": before_revoking_time},
        )
        print("Interval before revocation")
        print(alice_pres.verified, "should be true")

        # Request proof, no interval
        bob_pres, alice_pres = await indy_present_proof_v2(
            bob,
            alice,
            bob_conn.connection_id,
            alice_conn.connection_id,
            requested_attributes=[
                {
                    "name": "name",
                    "restrictions": [{"cred_def_id": cred_def_id}],
                }
            ],
        )
        print("No interval")
        print(alice_pres.verified, "should be true")

        # Request proof, using invalid/revoked interval but using
        # local non_revoked override (in requsted attrs)
        # ("LOCAL"-->requested attrs)
        bob_pres, alice_pres = await indy_present_proof_v2(
            bob,
            alice,
            bob_conn.connection_id,
            alice_conn.connection_id,
            requested_attributes=[
                {
                    "name": "name",
                    "restrictions": [{"cred_def_id": cred_def_id}],
                    "non_revoked": {
                        "from": before_revoking_time,
                        "to": before_revoking_time,
                    },
                }
            ],
            non_revoked={"from": revoked_time, "to": revoked_time},
        )
        print("Local interval overriding global?")
        print(alice_pres.verified, "should be true")

        # Request proof, just local invalid interval
        bob_pres, alice_pres = await indy_present_proof_v2(
            bob,
            alice,
            bob_conn.connection_id,
            alice_conn.connection_id,
            requested_attributes=[
                {
                    "name": "name",
                    "restrictions": [{"cred_def_id": cred_def_id}],
                    "non_revoked": {
                        "from": revoked_time,
                        "to": revoked_time,
                    },
                }
            ],
        )
        print("Local interval")
        print(alice_pres.verified, "should be false")


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

docker-compose.yml

version: '3'
services:
  alice:
    platform: linux/amd64
    image: acapy-run
    build:
      context: .
      dockerfile: ./docker/Dockerfile.run
    ports:
      - 3001:3001
    volumes:
      - ./aries_cloudagent:/usr/src/app/aries_cloudagent:z
    command: >
      start
        -it http 0.0.0.0 3000 -ot http -e http://alice:3000
        --admin 0.0.0.0 3001 --admin-insecure-mode
        --wallet-type askar
        --wallet-name default
        --wallet-key test
        --auto-provision
        --genesis-url https://raw.githubusercontent.com/Indicio-tech/indicio-network/main/genesis_files/pool_transactions_testnet_genesis
        --tails-server-base-url http://tails:6543
        --notify-revocation
        --debug-webhooks
        --log-level debug
    healthcheck:
      test: ["CMD-SHELL", "python", "healthcheck.py", "http://localhost:3001/status/live"]
      start_period: 10s
      interval: 3s
      timeout: 5s
      retries: 5
    depends_on:
      tails:
        condition: service_started

  bob:
    platform: linux/amd64
    image: acapy-run
    ports:
      - 3005:3005
    volumes:
      - ./aries_cloudagent:/usr/src/app/aries_cloudagent:z
    command: >
      start
        -it http 0.0.0.0 3004 -ot http -e http://bob:3004
        --admin 0.0.0.0 3005 --admin-insecure-mode
        --wallet-type askar
        --wallet-name default
        --wallet-key test
        --auto-provision
        --genesis-url https://raw.githubusercontent.com/Indicio-tech/indicio-network/main/genesis_files/pool_transactions_testnet_genesis
        --tails-server-base-url http://tails:6543
        --monitor-revocation-notification
        --debug-webhooks
        --log-level debug
    healthcheck:
      test: ["CMD-SHELL", "python", "healthcheck.py", "http://localhost:3005/status/live"]
      start_period: 10s
      interval: 3s
      timeout: 5s
      retries: 5
    depends_on:
      tails:
        condition: service_started

  tails:
    platform: linux/amd64
    image: ghcr.io/indicio-tech/tails-server:sha-3d2feb2
    ports:
      - 6543:6543
    environment:
      - GENESIS_URL=https://raw.githubusercontent.com/Indicio-tech/indicio-network/main/genesis_files/pool_transactions_testnet_genesis
    command: >
      tails-server
      --host 0.0.0.0
      --port 6543
      --storage-path /tmp/tails-files
      --log-level INFO

  tests:
    platform: linux/amd64
    image: anoncreds-test
    build:
      context: .
      dockerfile: ./docker/Dockerfile.anoncreds-test
    environment:
      ALICE: "http://alice:3001"
      BOB: "http://bob:3005"
    volumes:
      - ./anoncreds_test.py:/usr/src/app/anoncreds_test.py:z
    depends_on:
      alice:
        condition: service_healthy
      bob:
        condition: service_healthy

Dockerfile.anoncreds-test

FROM python:3.9-slim
WORKDIR /usr/src/app

RUN apt-get update && apt-get install -y git
RUN pip install git+https://github.com/Indicio-tech/acapy-minimal-example.git

ADD anoncreds_test.py .
CMD ["python", "anoncreds_test.py"]

healthcheck.py

import sys
from urllib.request import urlopen
from urllib.error import URLError, HTTPError


def check_url(url):
    try:
        with urlopen(url) as response:
            return response.status == 200
    except HTTPError as e:
        print(f"Error: {e}")
        return False
    except URLError as e:
        print(f"Error: {e}")
        return False


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python healthcheck.py <url>")
        sys.exit(1)

    url = sys.argv[1]
    is_healthy = check_url(url)
    sys.exit(0 if is_healthy else 1)