improve routes and did registration process

Signed-off-by: PatStLouis <[email protected]>

.dockerignore

@@ -1,2 +1,4 @@
 .env*
-__pycache__/
+*cache*
+*.db
+venv/

docker/Dockerfile → server/Dockerfile

@@ -1,14 +1,17 @@
 FROM python:3.12
 
-WORKDIR /trustdidweb-server-py
+WORKDIR /fastapi
 
 ENV VIRTUAL_ENV=/opt/venv
 RUN python3 -m venv $VIRTUAL_ENV
 ENV PATH="$VIRTUAL_ENV/bin:$PATH"
 
-COPY server/* ./
+COPY requirements.txt ./
 
 RUN pip install --upgrade pip
 RUN pip install -r requirements.txt
 
+COPY app ./app
+COPY config.py main.py ./
+
 CMD [ "python", "main.py" ]

server/app/__init__.py

@@ -0,0 +1,27 @@
+from fastapi import FastAPI, APIRouter
+from fastapi.responses import JSONResponse
+from app.routers import identifiers
+from config import settings
+from app.plugins import AskarStorage
+
+app = FastAPI(title=settings.PROJECT_TITLE, version=settings.PROJECT_VERSION)
+
+
+api_router = APIRouter()
+
+
+@api_router.get("/server/status", tags=["Server"], include_in_schema=False)
+async def server_status():
+    return JSONResponse(status_code=200, content={"status": "ok"})
+
+
+@api_router.get("/.well-known/did.json", tags=["Server"], include_in_schema=False)
+async def get_main_did_document():
+    did_document = await AskarStorage().fetch("didDocument", settings.DID_WEB_BASE)
+    return JSONResponse(status_code=200, content=did_document)
+
+
+api_router.include_router(identifiers.router, tags=["Identifiers"])
+
+
+app.include_router(api_router)

server/app/dependencies.py

@@ -1,44 +1,13 @@
 from fastapi import HTTPException
 from config import settings
-from app.models.web_requests import RegisterDID
-from app.plugins import AskarVerifier, AskarStorage
-from app.utilities import find_key, find_proof
+from app.plugins import AskarStorage
 
 
 async def identifier_available(namespace: str, identifier: str):
-    if await AskarStorage().fetch("didDocument", f"{namespace}:{identifier}"):
+    if await AskarStorage().fetch("didDocument", f"{settings.DID_WEB_BASE}:{namespace}:{identifier}"):
         raise HTTPException(status_code=409, detail="Identifier unavailable.")
 
 
 async def did_document_exists(namespace: str, identifier: str):
-    if not await AskarStorage().fetch("didDocument", f"{namespace}:{identifier}"):
+    if not await AskarStorage().fetch("didDocument", f"{settings.DID_WEB_BASE}:{namespace}:{identifier}"):
         raise HTTPException(status_code=404, detail="Ressource not found.")
-
-
-async def valid_did_registration(
-    namespace: str, identifier: str, request_body: RegisterDID
-):
-    document = request_body.model_dump(by_alias=True, exclude_none=True)["didDocument"]
-    proofs = document.pop("proof")
-    try:
-        assert (
-            document["id"] == f"{settings.DID_WEB_BASE}:{namespace}:{identifier}"
-        ), "Id mismatch between DID Document and requested endpoint."
-        assert (
-            len(document["verificationMethod"]) >= 1
-        ), "DID Documentmust contain at least 1 verificationMethod."
-        assert (
-            isinstance(proofs, list) and len(proofs) == 2
-        ), "Insuficient proofs, must contain a client and an endorser proof."
-    except AssertionError as msg:
-        raise HTTPException(status_code=400, detail=str(msg))
-
-    endorser_proof = find_proof(proofs, f"{settings.DID_WEB_BASE}#key-01")
-    endorser_key = settings.ENDORSER_MULTIKEY
-    AskarVerifier(endorser_key).verify_proof(document, endorser_proof)
-
-    client_proof = find_proof(proofs, document["verificationMethod"][0]["id"])
-    client_key = find_key(document, client_proof["verificationMethod"])
-    AskarVerifier(client_key).verify_proof(document, client_proof)
-
-    return document

server/app/models/di_proof.py

@@ -1,5 +1,10 @@
-from typing import Union, List, Dict
-from pydantic import BaseModel, Field, AliasChoices, field_validator
+from typing import Dict, Any
+from pydantic import BaseModel, Field, field_validator
+
+
+class BaseModel(BaseModel):
+    def model_dump(self, **kwargs) -> Dict[str, Any]:
+        return super().model_dump(by_alias=True, exclude_none=True, **kwargs)
 
 
 class DataIntegrityProof(BaseModel):
@@ -8,10 +13,10 @@ class DataIntegrityProof(BaseModel):
     verification_method: str = Field(alias="verificationMethod")
     proof_value: str = Field(alias="proofValue")
     proof_purpose: str = Field(alias="proofPurpose")
-    domain: str = Field()
-    challenge: str = Field()
-    created: str = Field()
-    expires: str = Field()
+    domain: str = Field(None)
+    challenge: str = Field(None)
+    created: str = Field(None)
+    expires: str = Field(None)
 
     @field_validator("type")
     @classmethod
@@ -26,3 +31,8 @@ def validate_cryptosuite(cls, value):
         if value not in ["eddsa-jcs-2022"]:
             raise ValueError("Unsupported cryptosuite")
         return value
+
+    @field_validator("expires")
+    @classmethod
+    def validate_expires(cls, value):
+        return value

server/app/models/did_document.py

@@ -1,6 +1,12 @@
-from typing import Union, List, Dict
-from pydantic import BaseModel, Field
+from typing import Union, List, Dict, Any
+from pydantic import BaseModel, Field, field_validator
 from .di_proof import DataIntegrityProof
+import re
+
+
+class BaseModel(BaseModel):
+    def model_dump(self, **kwargs) -> Dict[str, Any]:
+        return super().model_dump(by_alias=True, exclude_none=True, **kwargs)
 
 
 class VerificationMethod(BaseModel):
@@ -20,14 +26,56 @@ class DidDocument(BaseModel):
     context: Union[str, List[str]] = Field(
         ["https://www.w3.org/ns/did/v1"], alias="@context"
     )
-    id: str = Field(None)
+    id: str = Field()
     controller: str = Field(None)
     alsoKnownAs: List[str] = Field(None)
-    verificationMethod: List[VerificationMethod] = Field(None)
-    authentication: List[Union[str, VerificationMethod]] = Field(None)
-    assertionMethod: List[Union[str, VerificationMethod]] = Field(None)
+    verificationMethod: List[VerificationMethod] = Field()
+    authentication: List[Union[str, VerificationMethod]] = Field()
+    assertionMethod: List[Union[str, VerificationMethod]] = Field()
     keyAgreement: List[Union[str, VerificationMethod]] = Field(None)
     capabilityInvocation: List[Union[str, VerificationMethod]] = Field(None)
     capabilityDelegation: List[Union[str, VerificationMethod]] = Field(None)
     service: List[Service] = Field(None)
-    proof: Union[DataIntegrityProof, List[DataIntegrityProof]] = Field(None)
+
+    @field_validator("context")
+    @classmethod
+    def validate_context(cls, value):
+        assert value[0] == "https://www.w3.org/ns/did/v1", "Invalid context."
+        return value
+
+    @field_validator("id")
+    @classmethod
+    def validate_id(cls, value):
+        DID_REGEX = re.compile(
+            "did:([a-z0-9]+):((?:[a-zA-Z0-9._%-]*:)*[a-zA-Z0-9._%-]+)"
+        )
+        assert DID_REGEX.match(value), "Expected id to be a DID."
+        return value
+
+    @field_validator("authentication")
+    @classmethod
+    def validate_authentication(cls, value):
+        assert len(value) >= 1, "Expected at least one authentication method."
+        return value
+
+    @field_validator("assertionMethod")
+    @classmethod
+    def validate_assertion_method(cls, value):
+        assert len(value) >= 1, "Expected at least one assertion method."
+        return value
+
+    @field_validator("verificationMethod")
+    @classmethod
+    def validate_verification_method(cls, value):
+        assert len(value) >= 1, "Expected at least one verification method."
+        return value
+
+
+class SecuredDidDocument(DidDocument):
+    proof: List[DataIntegrityProof] = Field(None)
+
+    @field_validator("proof")
+    @classmethod
+    def validate_proof(cls, value):
+        assert len(value) == 2, "Expected proof set."
+        return value

server/app/models/web_schemas.py

@@ -0,0 +1,12 @@
+from typing import Dict, Any
+from pydantic import BaseModel, Field
+from .did_document import SecuredDidDocument
+
+
+class BaseModel(BaseModel):
+    def model_dump(self, **kwargs) -> Dict[str, Any]:
+        return super().model_dump(by_alias=True, exclude_none=True, **kwargs)
+
+
+class RegisterDID(BaseModel):
+    didDocument: SecuredDidDocument = Field()

server/app/plugins/askar.py

@@ -1,6 +1,6 @@
 import json
 from fastapi import HTTPException
-from aries_askar import Store, error, Key
+from aries_askar import Store, Key
 from aries_askar.bindings import LocalKeyHandle
 from config import settings
 from app.utilities import create_did_doc
@@ -78,8 +78,7 @@ def __init__(self, multikey=None):
                 alg="ed25519", public=bytes(bytearray(multibase.decode(multikey))[2:])
             )
 
-    def create_proof_config(self):
-        created = str(datetime.now(timezone.utc).isoformat("T", "seconds"))
+    def create_proof_config(self, did):
         expires = str(
             (datetime.now(timezone.utc) + timedelta(minutes=10)).isoformat(
                 "T", "seconds"
@@ -89,45 +88,46 @@ def create_proof_config(self):
             "type": self.type,
             "cryptosuite": self.cryptosuite,
             "proofPurpose": self.purpose,
-            "created": created,
+            "verificationMethod": f"{settings.DID_WEB_BASE}#key-01",
             "expires": expires,
             "domain": settings.DOMAIN,
-            "challenge": self.create_challenge(created + expires),
+            "challenge": self.create_challenge(did + expires),
         }
 
     def create_challenge(self, value):
         return str(uuid.uuid5(uuid.NAMESPACE_DNS, settings.SECRET_KEY + value))
 
-    def assert_proof_options(self, proof):
+    def assert_proof_options(self, proof, did):
         try:
+            assert datetime.fromisoformat(proof["expires"]) > datetime.now(
+                timezone.utc
+            ), "Proof expired."
+            assert proof["domain"] == settings.DOMAIN, "Domain mismatch."
+            assert proof["challenge"] == self.create_challenge(
+                did + proof["expires"]
+            ), "Challenge mismatch."
             assert proof["type"] == self.type, f"Expected {self.type} proof type."
             assert (
                 proof["cryptosuite"] == self.cryptosuite
             ), f"Expected {self.cryptosuite} proof cryptosuite."
             assert (
                 proof["proofPurpose"] == self.purpose
             ), f"Expected {self.purpose} proof purpose."
-            assert proof["domain"] == settings.DOMAIN, "Domain mismatch."
-            assert proof["challenge"] == self.create_challenge(
-                proof["created"] + proof["expires"]
-            ), "Challenge mismatch."
-            assert datetime.fromisoformat(proof["created"]) < datetime.now(
-                timezone.utc
-            ), "Invalid proof creation timestamp."
-            assert datetime.fromisoformat(proof["expires"]) > datetime.now(
-                timezone.utc
-            ), "Proof expired."
-            assert datetime.fromisoformat(proof["created"]) < datetime.fromisoformat(
-                proof["expires"]
-            ), "Proof validity period invalid."
         except AssertionError as msg:
             raise HTTPException(status_code=400, detail=str(msg))
 
     def verify_proof(self, document, proof):
-        self.assert_proof_options(proof)
-        assert (
-            proof["verificationMethod"].split("#")[0] == document["id"]
-            or proof["verificationMethod"].split("#")[0] == settings.DID_WEB_BASE
+        self.assert_proof_options(proof, document['id'])
+
+        # Set multikey to endorser multikey if verificationMethod matches else expect did:key:
+        multikey = (
+            settings.ENDORSER_MULTIKEY
+            if proof["verificationMethod"] == f'{settings.DID_WEB_BASE}#key-01'
+            else proof["verificationMethod"].split("#")[-1]
+        )
+
+        self.key = Key(LocalKeyHandle()).from_public_bytes(
+            alg="ed25519", public=bytes(bytearray(multibase.decode(multikey))[2:])
         )
 
         proof_options = proof.copy()