Merge pull request #6 from lalithkota/15.0-develop

Moved unittests from 17.0 branch

.github/workflows/test.yml

@@ -37,7 +37,7 @@ jobs:
     steps:
     - id: set-matrix
       run: |
-        BRANCH_NAME=$(echo ${{ github.ref }} | sed -e 's,.*/\(.*\),\1,')
+        BRANCH_NAME=$(echo ${{ github.base_ref || github.ref }} | sed -e 's,.*/\(.*\),\1,')
         if [[ $BRANCH_NAME = 15.0* ]] ; then
           matrix='{"include": [{
               "container": "ghcr.io/oca/oca-ci/py3.8-odoo15.0:latest",
@@ -57,7 +57,7 @@ jobs:
               "name": "test with OCB 17"
             }]}'
         fi
-        echo "matrix<<EOF"$'\n'"$matrix"$'\n'EOF >> $GITHUB_OUTPUT
+        echo "matrix=$matrix" | tr -d '\n' >> $GITHUB_OUTPUT
   test:
     needs: matrix_prep
     runs-on: ubuntu-22.04

g2p_openid_vci/data/default_credential_format.jq

@@ -22,33 +22,33 @@
                 "value": (.partner.name // null)
             }
         ],
-        "gender": [
+        "gender": (if .partner.gender then [
             {
                 "language": "eng",
-                "value": (.partner.gender // null)
+                "value": .partner.gender
             }
-        ],
+        ] else null end),
         "dateOfBirth": (.partner.birthdate // null),
         "email": (.partner.email // null),
         "phone": (.partner.phone // null),
-        "addressLine1": [
+        "addressLine1": (if .partner_address.street_address then [
             {
                 "language": "eng",
                 "value": .partner_address.street_address
             }
-        ],
-        "province": [
+        ] else null end),
+        "province": (if .partner_address.locality then [
             {
                 "language": "eng",
                 "value": .partner_address.locality
             }
-        ],
-        "region": [
+        ] else null end),
+        "region": (if .partner_address.region then [
             {
                 "language": "eng",
                 "value": .partner_address.region
             }
-        ],
+        ] else null end),
         "postalCode": .partner_address.postal_code,
         "face": .partner_face,
         "UIN": .reg_ids["NATIONAL ID"]?.value

g2p_openid_vci/models/vci_issuer.py

@@ -2,6 +2,7 @@
 
 import json
 import logging
+import os
 import uuid
 from datetime import datetime
 
@@ -11,7 +12,8 @@
 from jose import jwt
 from pyld import jsonld
 
-from odoo import api, fields, models, modules, tools
+from odoo import api, fields, models, tools
+from odoo.tools import misc
 
 from ..json_encoder import VCJSONEncoder
 
@@ -56,6 +58,9 @@ def issue_vc(self, credential_request: dict, token: str):
         # TODO: Raise better errors and error types
         auth_claims_unverified = jwt.get_unverified_claims(token)
         auth_scopes = auth_claims_unverified.get("scope", "").split()
+        auth_aud = auth_claims_unverified.get("aud", "")
+        if isinstance(auth_aud, str):
+            auth_aud = auth_aud.split()
 
         request_format = credential_request["format"]
         request_types = credential_request["credential_definition"]["type"]
@@ -94,18 +99,11 @@ def issue_vc(self, credential_request: dict, token: str):
                 issuer=auth_allowed_iss,
                 options={"verify_aud": False},
             )
-            if auth_allowed_aud and (
-                (
-                    isinstance(auth_claims_unverified["aud"], list)
-                    and set(auth_allowed_aud).issubset(set(auth_claims_unverified["aud"]))
-                )
-                or (
-                    isinstance(auth_claims_unverified["aud"], str)
-                    and auth_allowed_aud in auth_claims_unverified["aud"]
-                )
-            ):
+            if auth_allowed_aud and not set(auth_allowed_aud).issubset(set(auth_aud)):
                 raise ValueError("Invalid Audience")
         except Exception as e:
+            if isinstance(e, ValueError) and "Invalid Audience" in str(e):
+                raise e
             raise ValueError("Invalid Auth Token received") from e
 
         issue_vc_func = getattr(credential_issuer, f"issue_vc_{credential_issuer.issuer_type}")
@@ -200,6 +198,57 @@ def build_empty_ld_proof(self):
             "proofPurpose": "assertionMethod",
         }
 
+    @api.model
+    def get_issuer_metadata_by_name(self, issuer_name=""):
+        """
+        If issuer_name param is null, this returns all issuer's metdata.
+        """
+        search_domain = []
+        if issuer_name:
+            search_domain.append(("name", "=", issuer_name))
+        vci_issuers = self.sudo().search(search_domain)
+        return vci_issuers.get_issuer_metadata()
+
+    def get_issuer_metadata(self):
+        vci_issuers = self.read()
+        web_base_url = self.env["ir.config_parameter"].sudo().get_param("web.base.url").rstrip("/")
+        cred_configs = None
+        for issuer in vci_issuers:
+            issuer["web_base_url"] = web_base_url
+            issuer_metadata = jq.first(
+                issuer["issuer_metadata_text"], VCJSONEncoder.python_dict_to_json_dict(issuer)
+            )
+            if isinstance(issuer_metadata, list):
+                if not cred_configs:
+                    cred_configs = []
+                cred_configs.extend(issuer_metadata)
+            elif isinstance(issuer_metadata, dict):
+                if not cred_configs:
+                    cred_configs = {}
+                cred_configs.update(issuer_metadata)
+        response = {
+            "credential_issuer": web_base_url,
+            "credential_endpoint": f"{web_base_url}/api/v1/vci/credential",
+        }
+        if isinstance(cred_configs, list):
+            response["credentials_supported"] = cred_configs
+        elif isinstance(cred_configs, dict):
+            response["credential_configurations_supported"] = cred_configs
+        return response
+
+    @api.model
+    def get_all_contexts_json(self):
+        web_base_url = self.env["ir.config_parameter"].sudo().get_param("web.base.url").rstrip("/")
+        context_jsons = self.sudo().search([]).read(["contexts_json"])
+        final_context = {"@context": {}}
+        for context in context_jsons:
+            context = context["contexts_json"].strip()
+            if context:
+                final_context["@context"].update(
+                    json.loads(context.replace("web_base_url", web_base_url))["@context"]
+                )
+        return final_context
+
     def get_auth_jwks(
         self,
         auth_issuer: str,
@@ -280,25 +329,23 @@ def set_default_credential_type_Registry(self):
     def set_from_static_file_Registry(
         self, module_name="g2p_openid_vci", file_name="", field_name="", **kwargs
     ):
-        default_path = modules.get_resource_path(module_name, "data", file_name)
         text = ""
         try:
-            with open(default_path) as file:
+            with misc.file_open(os.path.join(module_name, "data", file_name)) as file:
                 text = file.read()
                 if field_name:
                     self.write({field_name: text})
         except Exception:
             _logger.exception(f"Could not set default {field_name}")
         return text
 
-    @api.model
-    def verify_proof_and_bind(self, credential_request):
-        # TODO: Verify proof and do wallet binding
-        # request_proof_type = credential_request["proof"]["proof_type"]
-        # request_proof_jwt = credential_request["proof"]["jwt"]
-        # request_proof = None
-        # if request_proof_type and request_proof_jwt and request_proof_type == "jwt":
-        #     request_proof = jwt.get_unverified_claims(request_proof_jwt)
-        # else:
-        #     raise ValueError("Only JWT proof supported")
-        pass
+    # TODO: Verify proof and do wallet binding
+    # @api.model
+    # def verify_proof_and_bind(self, credential_request):
+    #     request_proof_type = credential_request["proof"]["proof_type"]
+    #     request_proof_jwt = credential_request["proof"]["jwt"]
+    #     request_proof = None
+    #     if request_proof_type and request_proof_jwt and request_proof_type == "jwt":
+    #         request_proof = jwt.get_unverified_claims(request_proof_jwt)
+    #     else:
+    #         raise ValueError("Only JWT proof supported")

g2p_openid_vci/tests/__init__.py

@@ -0,0 +1 @@
+from . import test_openid_vc_issuer