Simple decoupled file based policy engine

README.md

@@ -138,6 +138,7 @@ The `service_parameters.json` file gets created when starting a service using `.
     "serviceId": "emulator",
     "treeAlgorithm": "CCF",
     "signatureAlgorithm": "ES256",
+    "insertPolicy": "*",
     "serviceCertificate": "-----BEGIN CERTIFICATE-----..."
 }
 ```

dev-requirements.txt

@@ -1,5 +1,7 @@
 pyOpenSSL
 pytest
+jsonschema
 requests==2.31.0
 requests-toolbelt==0.9
 urllib3<2.0.0
+myst-parser

docs/registration_policies.md

@@ -0,0 +1,183 @@
+# Registration Policies
+
+- References
+  - [5.2.2. Registration Policies](https://www.ietf.org/archive/id/draft-birkholz-scitt-architecture-02.html#name-registration-policies)
+
+## Simple decoupled file based policy engine
+
+The SCITT API emulator can deny entry based on presence of
+`operation.policy.{insert,denied,failed}` files. Currently only for use with
+`use_lro=True`.
+
+This is a simple way to enable evaluation of claims prior to submission by
+arbitrary policy engines which watch the workspace (fanotify, inotify, etc.).
+
+[![asciicast-of-simple-decoupled-file-based-policy-engine](https://asciinema.org/a/572766.svg)](https://asciinema.org/a/572766)
+
+Start the server
+
+```console
+$ rm -rf workspace/
+$ mkdir -p workspace/storage/operations
+$ scitt-emulator server --workspace workspace/ --tree-alg CCF --use-lro
+Service parameters: workspace/service_parameters.json
+^C
+```
+
+Modification of config to non-`*` insert policy. Restart SCITT API emulator server after this.
+
+```console
+$ echo "$(cat workspace/service_parameters.json)" \
+    | jq '.insertPolicy = "allowlist.schema.json"' \
+    | tee workspace/service_parameters.json.new \
+    && mv workspace/service_parameters.json.new workspace/service_parameters.json
+{
+  "serviceId": "emulator",
+  "treeAlgorithm": "CCF",
+  "signatureAlgorithm": "ES256",
+  "serviceCertificate": "-----BEGIN CERTIFICATE-----",
+  "insertPolicy": "allowlist.schema.json"
+}
+```
+
+Basic policy engine in two files
+
+**enforce_policy.py**
+
+```python
+import os
+import sys
+import pathlib
+
+policy_reason = ""
+if "POLICY_REASON_PATH" in os.environ:
+    policy_reason = pathlib.Path(os.environ["POLICY_REASON_PATH"]).read_text()
+
+cose_path = pathlib.Path(sys.argv[-1])
+policy_action_path = cose_path.with_suffix(".policy." + os.environ["POLICY_ACTION"].lower())
+policy_action_path.write_text(policy_reason)
+```
+
+Simple drop rule based on claim content allowlist.
+
+**allowlist.schema.json**
+
+```json
+{
+    "$id": "https://schema.example.com/scitt-allowlist.schema.json",
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "properties": {
+        "issuer": {
+            "type": "string",
+            "enum": [
+                "did:web:example.org"
+            ]
+        }
+    }
+}
+```
+
+**jsonschema_validator.py**
+
+```python
+import os
+import sys
+import json
+import pathlib
+import traceback
+
+import cbor2
+import pycose
+from jsonschema import validate, ValidationError
+from pycose.messages import CoseMessage, Sign1Message
+
+from scitt_emulator.scitt import ClaimInvalidError, COSE_Headers_Issuer
+
+claim = sys.stdin.buffer.read()
+
+msg = CoseMessage.decode(claim)
+
+if pycose.headers.ContentType not in msg.phdr:
+    raise ClaimInvalidError("Claim does not have a content type header parameter")
+if COSE_Headers_Issuer not in msg.phdr:
+    raise ClaimInvalidError("Claim does not have an issuer header parameter")
+
+if not msg.phdr[pycose.headers.ContentType].startswith("application/json"):
+    raise TypeError(
+        f"Claim content type does not start with application/json: {msg.phdr[pycose.headers.ContentType]!r}"
+    )
+
+SCHEMA = json.loads(pathlib.Path(os.environ["SCHEMA_PATH"]).read_text())
+
+try:
+    validate(
+        instance={
+            "$schema": "https://schema.example.com/scitt-policy-engine-jsonschema.schema.json",
+            "issuer": msg.phdr[COSE_Headers_Issuer],
+            "claim": json.loads(msg.payload.decode()),
+        },
+        schema=SCHEMA,
+    )
+except ValidationError as error:
+    print(str(error), file=sys.stderr)
+    sys.exit(1)
+```
+
+We'll create a small wrapper to serve in place of a more fully featured policy
+engine.
+
+**policy_engine.sh**
+
+```bash
+export SCHEMA_PATH="${1}"
+CLAIM_PATH="${2}"
+
+echo ${CLAIM_PATH}
+
+(python3 jsonschema_validator.py < ${CLAIM_PATH} 2>error && POLICY_ACTION=insert python3 enforce_policy.py ${CLAIM_PATH}) || (python3 -c 'import sys, json; print(json.dumps({"type": "denied", "detail": sys.stdin.read()}))' < error > reason.json; POLICY_ACTION=denied POLICY_REASON_PATH=reason.json python3 enforce_policy.py ${CLAIM_PATH})
+```
+
+Example running allowlist check and enforcement.
+
+```console
+$ npm install -g nodemon
+$ nodemon -e .cose --exec 'find workspace/storage/operations -name \*.cose -exec nohup sh -xe policy_engine.sh $(cat workspace/service_parameters.json | jq -r .insertPolicy) {} \;'
+```
+
+Also ensure you restart the server with the new config we edited.
+
+```console
+$ scitt-emulator server --workspace workspace/ --tree-alg CCF --use-lro
+```
+
+Create claim from allowed issuer (`.org`) and from non-allowed (`.com`).
+
+```console
+$ scitt-emulator client create-claim --issuer did:web:example.com --content-type application/json --payload '{"sun": "yellow"}' --out claim.cose
+Claim written to claim.cose
+$ scitt-emulator client submit-claim --claim claim.cose --out claim.receipt.cbor
+Traceback (most recent call last):
+  File "/home/alice/.local/bin/scitt-emulator", line 33, in <module>
+    sys.exit(load_entry_point('scitt-emulator', 'console_scripts', 'scitt-emulator')())
+  File "/home/alice/Documents/python/scitt-api-emulator/scitt_emulator/cli.py", line 22, in main
+    args.func(args)
+  File "/home/alice/Documents/python/scitt-api-emulator/scitt_emulator/client.py", line 196, in <lambda>
+    func=lambda args: submit_claim(
+  File "/home/alice/Documents/python/scitt-api-emulator/scitt_emulator/client.py", line 107, in submit_claim
+    raise_for_operation_status(operation)
+  File "/home/alice/Documents/python/scitt-api-emulator/scitt_emulator/client.py", line 43, in raise_for_operation_status
+    raise ClaimOperationError(operation)
+scitt_emulator.client.ClaimOperationError: Operation error denied: 'did:web:example.com' is not one of ['did:web:example.org']
+
+Failed validating 'enum' in schema['properties']['issuer']:
+    {'enum': ['did:web:example.org'], 'type': 'string'}
+
+On instance['issuer']:
+    'did:web:example.com'
+
+$ scitt-emulator client create-claim --issuer did:web:example.org --content-type application/json --payload '{"sun": "yellow"}' --out claim.cose
+Claim written to claim.cose
+$ scitt-emulator client submit-claim --claim claim.cose --out claim.receipt.cbor
+Claim registered with entry ID 1
+Receipt written to claim.receipt.cbor
+```

environment.yml

@@ -34,3 +34,5 @@ dependencies:
     - rkvst-archivist==0.20.0
     - six==1.16.0
     - urllib3<2.0.0
+    - myst-parser==1.0.0
+    - jsonschema==4.17.3

scitt_emulator/client.py

@@ -17,6 +17,20 @@
 HTTP_DEFAULT_RETRY_DELAY = 1
 
 
+class ClaimOperationError(Exception):
+    def __init__(self, operation):
+        self.operation = operation
+
+    def __str__(self):
+        error_type = self.operation.get("error", {}).get(
+            "type", "error.type not present",
+        )
+        error_detail = self.operation.get("error", {}).get(
+            "detail", "error.detail not present",
+        )
+        return f"Operation error {error_type}: {error_detail}"
+
+
 def raise_for_status(response: httpx.Response):
     if response.is_success:
         return
@@ -26,7 +40,7 @@ def raise_for_status(response: httpx.Response):
 def raise_for_operation_status(operation: dict):
     if operation["status"] != "failed":
         return
-    raise RuntimeError(f"Operation error: {operation['error']}")
+    raise ClaimOperationError(operation)
 
 
 class HttpClient:

scitt_emulator/scitt.py

@@ -4,6 +4,7 @@
 from typing import Optional
 from abc import ABC, abstractmethod
 from pathlib import Path
+import contextlib
 import time
 import json
 import uuid
@@ -22,6 +23,10 @@
 COSE_Headers_Tree_Alg = "tree_alg"
 COSE_Headers_Issued_At = "issued_at"
 
+# permissive insert policy
+MOST_PERMISSIVE_INSERT_POLICY = "*"
+DEFAULT_INSERT_POLICY = MOST_PERMISSIVE_INSERT_POLICY
+
 
 class ClaimInvalidError(Exception):
     pass
@@ -35,6 +40,10 @@ class OperationNotFoundError(Exception):
     pass
 
 
+class PolicyResultDecodeError(Exception):
+    pass
+
+
 class SCITTServiceEmulator(ABC):
     def __init__(
         self, service_parameters_path: Path, storage_path: Optional[Path] = None
@@ -94,8 +103,14 @@ def get_claim(self, entry_id: str) -> bytes:
         return claim
 
     def submit_claim(self, claim: bytes, long_running=True) -> dict:
+        insert_policy = self.service_parameters.get("insertPolicy", DEFAULT_INSERT_POLICY)
+
         if long_running:
             return self._create_operation(claim)
+        elif insert_policy != MOST_PERMISSIVE_INSERT_POLICY:
+            raise NotImplementedError(
+                f"non-* insertPolicy only works with long_running=True: {insert_policy!r}"
+            )
         else:
             return self._create_entry(claim)
 
@@ -142,11 +157,57 @@ def _create_operation(self, claim: bytes):
 
         return operation
 
+    def _sync_policy_result(self, operation: dict):
+        operation_id = operation["operationId"]
+        policy_insert_path = self.operations_path / f"{operation_id}.policy.insert"
+        policy_denied_path = self.operations_path / f"{operation_id}.policy.denied"
+        policy_failed_path = self.operations_path / f"{operation_id}.policy.failed"
+        insert_policy = self.service_parameters.get("insertPolicy", DEFAULT_INSERT_POLICY)
+
+        policy_result = {"status": operation["status"]}
+
+        if insert_policy == MOST_PERMISSIVE_INSERT_POLICY:
+            policy_result["status"] = "succeeded"
+        if policy_insert_path.exists():
+            policy_result["status"] = "succeeded"
+            policy_insert_path.unlink()
+        if policy_failed_path.exists():
+            policy_result["status"] = "failed"
+            if policy_failed_path.stat().st_size != 0:
+                try:
+                    policy_result_error = json.loads(policy_failed_path.read_text())
+                except Exception as error:
+                    raise PolicyResultDecodeError(operation_id) from error
+                policy_result["error"] = policy_result_error
+            policy_failed_path.unlink()
+        if policy_denied_path.exists():
+            policy_result["status"] = "denied"
+            if policy_denied_path.stat().st_size != 0:
+                try:
+                    policy_result_error = json.loads(policy_denied_path.read_text())
+                except Exception as error:
+                    raise PolicyResultDecodeError(operation_id) from error
+                policy_result["error"] = policy_result_error
+            policy_denied_path.unlink()
+
+        return policy_result
+
     def _finish_operation(self, operation: dict):
         operation_id = operation["operationId"]
         operation_path = self.operations_path / f"{operation_id}.json"
         claim_src_path = self.operations_path / f"{operation_id}.cose"
 
+        policy_result = self._sync_policy_result(operation)
+        if policy_result["status"] == "running":
+            return operation
+        if policy_result["status"] != "succeeded":
+            operation["status"] = "failed"
+            if "error" in policy_result:
+                operation["error"] = policy_result["error"]
+            operation_path.unlink()
+            claim_src_path.unlink()
+            return operation
+
         claim = claim_src_path.read_bytes()
         entry = self._create_entry(claim)
         claim_src_path.unlink()