Check Asset Compliance: SINCE policy

# Define a compliance policy that alerts when an asset has expired.

# Main function parses in a url to the Archivist and client credentials , which is
# a user authorization. The main function would initialize an archivist connection
# using the url and the credentials, called "arch", then call arch.access_policies.list()
# with suitable properties and attributes.

from json import dumps as json_dumps
from os import getenv
from time import sleep
from uuid import uuid4
from warnings import filterwarnings

from dotenv import load_dotenv

from archivist.archivist import Archivist
from archivist.compliance_policy_requests import (
    CompliancePolicySince,
)
from archivist.logger import set_logger

filterwarnings("ignore", message="Unverified HTTPS request")

%reload_ext dotenv
%dotenv -o notebooks.env

# URL, CLIENT, SECRET are environment variables that represent connection parameters.
#
# URL = represents the url to the DataTrails application
# CLIENT = represents the client ID from an Application Registration
# SECRET = represents the client secret from an Application Registration
DATATRAILS_URL = getenv("DATATRAILS_URL")
DATATRAILS_APPREG_CLIENT = getenv("DATATRAILS_APPREG_CLIENT")
DATATRAILS_APPREG_SECRET = getenv("DATATRAILS_APPREG_SECRET")

"""
Main function of Asset and Event creation.

* Connect to DataTrails with client ID and client secret
* Creates an Asset and two Events
* Prints response of Asset and Event creation
"""

# Optional call to set the logger level.  The argument can be either
# "INFO" or "DEBUG".  For more sophisticated logging control see our
# documentation.
set_logger("INFO")

# Initialize connection to DATATRAILS
print("Connecting to DATATRAILS")
print("DATATRAILS_URL", DATATRAILS_URL)
arch = Archivist(
    DATATRAILS_URL, (DATATRAILS_APPREG_CLIENT, DATATRAILS_APPREG_SECRET), max_time=300
)

Connecting to DATATRAILS
DATATRAILS_URL https://app.datatrails.ai

def create_compliance_policy(arch, tag):
    """Compliance policy which expires 10 seconds after a
    Maintenance Performed event on a 'Traffic Light' has occurred.

    Usually the expiry time is on the order of days or weeks..

    Additionally the use of tag is simply to make this example
    repeatable.
    """
    compliance_policy = arch.compliance_policies.create(
        CompliancePolicySince(
            description="Maintenance should be performed every 10 seconds",
            display_name="Regular Maintenance of Traffic light",
            asset_filter=[
                ["attributes.arc_display_type=Traffic Light"],
            ],
            event_display_type=f"Maintenance Performed {tag}",
            time_period_seconds=10,  # very short so we can test
        )
    )
    print("SINCE_POLICY:", json_dumps(compliance_policy, indent=4))
    return compliance_policy

def create_traffic_light(arch):
    """
    Creates a traffic light.

    Note that arc_display_type siginfies a Traffic Light
    """

    traffic_light = arch.assets.create(
        attrs={
            "arc_display_name": "Traffic light model 54",
            "arc_description": "Traffic flow control light at A603 North East",
            "arc_display_type": "Traffic Light",
        },
        confirm=True,
    )
    print("TRAFFIC_LIGHT:", json_dumps(traffic_light, indent=4))
    return traffic_light

def perform_maintenance(arch, traffic_light, tag):
    """
    Perform maintenance on traffic light
    """
    maintenance_performed = arch.events.create(
        traffic_light["identity"],
        {
            "operation": "Record",
            "behaviour": "RecordEvidence",
        },
        {
            "arc_description": "Maintenance performed on traffic light",
            "arc_display_type": f"Maintenance Performed {tag}",
        },
        confirm=True,
    )
    print("MAINTENANCE_PERFORMED:", json_dumps(maintenance_performed, indent=4))

tag = uuid4()  # make this example repeatable

# make a SINCE compliance policy that alerts when the
# maintenance performed event has expired.
compliance_policy = create_compliance_policy(arch, tag)
print("compliance_policy", json_dumps(compliance_policy, indent=4))

Refresh token

SINCE_POLICY: {
    "identity": "compliance_policies/458957bc-4da7-4cc3-b37f-43fa53abe0cc",
    "compliance_type": "COMPLIANCE_SINCE",
    "description": "Maintenance should be performed every 10 seconds",
    "display_name": "Regular Maintenance of Traffic light",
    "asset_filter": [
        {
            "or": [
                "attributes.arc_display_type=Traffic Light"
            ]
        }
    ],
    "event_display_type": "Maintenance Performed a3f86bbf-737a-45d8-bd84-3d6612fb641e",
    "closing_event_display_type": "",
    "time_period_seconds": "10",
    "dynamic_window": "0",
    "dynamic_variability": 0,
    "richness_assertions": []
}
compliance_policy {
    "identity": "compliance_policies/458957bc-4da7-4cc3-b37f-43fa53abe0cc",
    "compliance_type": "COMPLIANCE_SINCE",
    "description": "Maintenance should be performed every 10 seconds",
    "display_name": "Regular Maintenance of Traffic light",
    "asset_filter": [
        {
            "or": [
                "attributes.arc_display_type=Traffic Light"
            ]
        }
    ],
    "event_display_type": "Maintenance Performed a3f86bbf-737a-45d8-bd84-3d6612fb641e",
    "closing_event_display_type": "",
    "time_period_seconds": "10",
    "dynamic_window": "0",
    "dynamic_variability": 0,
    "richness_assertions": []
}

# create an asset that matches the assets_filter field in the
# compliance policy.
traffic_light = create_traffic_light(arch)
print("traffic_light", json_dumps(traffic_light, indent=4))

TRAFFIC_LIGHT: {
    "identity": "assets/b6f63a6d-24a1-4dd8-a6d7-21e50e603ceb",
    "behaviours": [
        "RecordEvidence",
        "Builtin",
        "AssetCreator",
    ],
    "attributes": {
        "arc_description": "Traffic flow control light at A603 North East",
        "arc_display_name": "Traffic light model 54",
        "arc_display_type": "Traffic Light"
    },
    "confirmation_status": "CONFIRMED",
    "tracked": "TRACKED",
    "owner": "0xe889E67FdBa658C6f27ccBDa98D9d1B5500Dbbce",
    "at_time": "2023-01-16T11:52:27Z",
    "storage_integrity": "TENANT_STORAGE",
    "proof_mechanism": "SIMPLE_HASH",
    "chain_id": "827586838445807967",
    "public": false,
    "tenant_identity": "tenant/9bfb80ee-81f6-40dc-b5c7-1c7fb2fb9866"
}
traffic_light {
    "identity": "assets/b6f63a6d-24a1-4dd8-a6d7-21e50e603ceb",
    "behaviours": [
        "RecordEvidence",
        "Builtin",
        "AssetCreator",
    ],
    "attributes": {
        "arc_description": "Traffic flow control light at A603 North East",
        "arc_display_name": "Traffic light model 54",
        "arc_display_type": "Traffic Light"
    },
    "confirmation_status": "CONFIRMED",
    "tracked": "TRACKED",
    "owner": "0xe889E67FdBa658C6f27ccBDa98D9d1B5500Dbbce",
    "at_time": "2023-01-16T11:52:27Z",
    "storage_integrity": "TENANT_STORAGE",
    "proof_mechanism": "SIMPLE_HASH",
    "chain_id": "827586838445807967",
    "public": false,
    "tenant_identity": "tenant/9bfb80ee-81f6-40dc-b5c7-1c7fb2fb9866"
}

# perform maintenance on the asset which is valid for 10 seconds.
perform_maintenance(arch, traffic_light, tag)

# and check compliance - should be OK.
print("Sleep 1 second...")
sleep(1)
compliance = arch.compliance.compliant_at(
    traffic_light["identity"],
)
print("COMPLIANCE (true):", json_dumps(compliance, indent=4))

MAINTENANCE_PERFORMED: {
    "identity": "assets/b6f63a6d-24a1-4dd8-a6d7-21e50e603ceb/events/eba5bb05-d4ff-4d99-9205-41236560d24d",
    "asset_identity": "assets/b6f63a6d-24a1-4dd8-a6d7-21e50e603ceb",
    "event_attributes": {
        "arc_description": "Maintenance performed on traffic light",
        "arc_display_type": "Maintenance Performed a3f86bbf-737a-45d8-bd84-3d6612fb641e"
    },
    "asset_attributes": {},
    "operation": "Record",
    "behaviour": "RecordEvidence",
    "timestamp_declared": "2023-01-16T11:52:31Z",
    "timestamp_accepted": "2023-01-16T11:52:31Z",
    "timestamp_committed": "2023-01-16T11:52:31.599813432Z",
    "principal_declared": {
        "issuer": "https://app.datatrails.ai/appidpv1",
        "subject": "437bd138-dade-4346-aadd-dfdfee51ddf4",
        "display_name": "Test Notebooks",
        "email": ""
    },
    "principal_accepted": {
        "issuer": "https://app.datatrails.ai/appidpv1",
        "subject": "437bd138-dade-4346-aadd-dfdfee51ddf4",
        "display_name": "Test Notebooks",
        "email": ""
    },
    "confirmation_status": "CONFIRMED",
    "transaction_id": "",
    "block_number": 0,
    "transaction_index": 0,
    "from": "0xe889E67FdBa658C6f27ccBDa98D9d1B5500Dbbce",
    "tenant_identity": "tenant/9bfb80ee-81f6-40dc-b5c7-1c7fb2fb9866"
}
Sleep 1 second...
COMPLIANCE (true): {
    "compliant": true,
    "compliance": [
        {
            "compliance_policy_identity": "compliance_policies/458957bc-4da7-4cc3-b37f-43fa53abe0cc",
            "compliant": true,
            "reason": ""
        }
    ],
    "next_page_token": "",
    "compliant_at": "2023-01-16T11:52:33Z"
}

# however waiting long enough (> 10s) will cause the asset to
# become non-compliant...
print("Sleep 15 seconds...")
sleep(15)
compliance = arch.compliance.compliant_at(
    traffic_light["identity"],
)
print("COMPLIANCE (false):", json_dumps(compliance, indent=4))

Sleep 15 seconds...
COMPLIANCE (false): {
    "compliant": false,
    "compliance": [
        {
            "compliance_policy_identity": "compliance_policies/458957bc-4da7-4cc3-b37f-43fa53abe0cc",
            "compliant": false,
            "reason": "Duration 20s exceeds limit 10s"
        }
    ],
    "next_page_token": "",
    "compliant_at": "2023-01-16T11:52:52Z"
}

# finally delete the compliance_policy
_ = arch.compliance_policies.delete(
    compliance_policy["identity"],
)