fix CWL namespace variable schema URI validation

Author: fmigneault

tests/wps_restapi/test_processes.py

@@ -41,6 +41,7 @@
 from weaver.formats import AcceptLanguage, ContentType, get_cwl_file_format
 from weaver.processes.builtin import register_builtin_processes
 from weaver.processes.constants import (
+    CWL_NAMESPACE_WEAVER,
     CWL_REQUIREMENT_APP_DOCKER,
     CWL_REQUIREMENT_APP_OGC_API,
     CWL_REQUIREMENT_APP_WPS1,
@@ -1545,7 +1546,7 @@ def validate_ogcapi_process_description(self, process_description, process_id, r
         ref = self.get_application_package(remote_process)
         pkg = self.get_application_package(process_id)
         assert pkg["hints"] == {
-            "OGCAPIRequirement": {
+            f"{CWL_NAMESPACE_WEAVER}:{CWL_REQUIREMENT_APP_OGC_API}": {
                 "process": ref_url
             }
         }

tests/wps_restapi/test_swagger_definitions.py

@@ -64,23 +64,52 @@ def test_process_id_with_version_tag_get_valid():
     {IANA_NAMESPACE: IANA_NAMESPACE_URL, CWL_NAMESPACE: CWL_NAMESPACE_URL},
     {CWL_NAMESPACE: CWL_NAMESPACE_URL, CWL_NAMESPACE_WEAVER: CWL_NAMESPACE_WEAVER_URL},
     {CWL_NAMESPACE: CWL_NAMESPACE_URL, "random": "https://random.com"},
+    {
+        CWL_NAMESPACE: CWL_NAMESPACE_URL,
+        "random": "https://random.com",
+        "another": "https://another.com#",  # ensure no '/' does not cause an error (fails with URL regex, needs '/#')
+        "slashed": "https://another.com/#",
+        "object": "https://another.com/schema#object",
+    },
 ])
-def test_cwl_namespaces(test_value):
+def test_cwl_namespaces_valid(test_value):
     result = sd.CWLNamespaces().deserialize(test_value)
     assert result == test_value
 
 
 @pytest.mark.parametrize("test_value", [
     {CWL_NAMESPACE: "bad"},
     {CWL_NAMESPACE: EDAM_NAMESPACE_URL},
+    {CWL_NAMESPACE_WEAVER: "https://random.com"},  # disallow conflict with well-known namespaces, even if URI is valid
     {"random": "bad"},
     {"random": 12345},
+    {"bad": "bad", "good": "https://random.com"},  # disallow partial mapping even if other URI are valid
 ])
-def test_cwl_namespaces_invalid_url(test_value):
+def test_cwl_namespaces_invalid_uri(test_value):
     with pytest.raises(colander.Invalid):
         sd.CWLNamespaces().deserialize(test_value)
 
 
+@pytest.mark.parametrize(["test_value", "expect_result"], [
+    # literal property name under class definition
+    ({"var": "bad"}, colander.Invalid),
+    ({"var": "https://random.com"}, {"var": "https://random.com"}),  # valid
+    # name under 'variable' keyword passed as argument
+    ({"{namespace}": "bad"}, colander.Invalid),
+    ({"{namespace}": "https://random.com"}, {"{namespace}": "https://random.com"}),  # valid
+])
+def test_cwl_namespaces_var_not_conflict_namespace_name(test_value, expect_result):
+    assert any(getattr(field, "variable", None) is not None for field in sd.CWLNamespaces().children), (
+        "Requirement not met for test. Field 'variable' expected to be defined in schema definition."
+    )
+    if expect_result is colander.Invalid:
+        with pytest.raises(colander.Invalid):
+            sd.CWLNamespaces().deserialize(test_value)
+    else:
+        result = sd.CWLNamespaces().deserialize(test_value)
+        assert result == expect_result
+
+
 @pytest.mark.parametrize(
     ["test_data", "expect_result"],
     [

weaver/datatype.py

@@ -36,7 +36,14 @@
 from weaver.exceptions import ProcessInstanceError, ServiceParsingError
 from weaver.execute import ExecuteControlOption, ExecuteMode, ExecuteResponse, ExecuteTransmissionMode
 from weaver.formats import AcceptLanguage, ContentType, repr_json
-from weaver.processes.constants import CWL_REQUIREMENT_APP_DOCKER, CWL_REQUIREMENT_APP_OGC_API, ProcessSchema
+from weaver.processes.constants import (
+    CWL_NAMESPACE_WEAVER,
+    CWL_REQUIREMENT_APP_DOCKER,
+    CWL_REQUIREMENT_APP_DOCKER_GPU,
+    CWL_REQUIREMENT_APP_OGC_API,
+    CWL_REQUIREMENT_APP_WPS1,
+    ProcessSchema
+)
 from weaver.processes.convert import get_field, json2oas_io, normalize_ordered_io, null, ows2json, wps2json_io
 from weaver.processes.types import ProcessType
 from weaver.quotation.status import QuoteStatus
@@ -2184,11 +2191,20 @@ def deployment_profile(self):
 
         if cls == ProcessType.WORKFLOW:
             profile = f"{base}workflow"
-        elif ProcessType.is_wps(typ):
+        elif ProcessType.is_wps(typ) or req in [
+            CWL_REQUIREMENT_APP_WPS1,
+            f"{CWL_NAMESPACE_WEAVER}:{CWL_REQUIREMENT_APP_WPS1}",
+        ]:
             profile = f"{base}wpsApplication"
-        elif typ == ProcessType.OGC_API or req == CWL_REQUIREMENT_APP_OGC_API:
+        elif typ == ProcessType.OGC_API or req in [
+            CWL_REQUIREMENT_APP_OGC_API,
+            f"{CWL_NAMESPACE_WEAVER}:{CWL_REQUIREMENT_APP_OGC_API}",
+        ]:
             profile = f"{base}ogcapiApplication"
-        elif typ == ProcessType.APPLICATION or req == CWL_REQUIREMENT_APP_DOCKER:
+        elif typ == ProcessType.APPLICATION or req in [
+            CWL_REQUIREMENT_APP_DOCKER,
+            CWL_REQUIREMENT_APP_DOCKER_GPU,
+        ]:
             profile = f"{base}dockerizedApplication"
         else:
             profile = base + typ

weaver/wps_restapi/colander_extras.py

@@ -51,6 +51,7 @@
     of generated OpenAPI model definitions. If not explicitly provided, the
     value of ``title`` **WILL** default to the name of the schema node class.
 """
+import copy
 import inspect
 import re
 import uuid
@@ -134,7 +135,7 @@
 URL = colander.Regex(URL_REGEX, msg=colander._("Must be a URL"), flags=re.IGNORECASE)
 FILE_URL_REGEX = colander.URI_REGEX.replace(r"://", r"://(?!//)")
 FILE_URI = colander.Regex(FILE_URL_REGEX, msg=colander._("Must be a file:// URI scheme"), flags=re.IGNORECASE)
-URI_REGEX = rf"{colander.URL_REGEX[:-1]}(?:#?|[#?]\S+)$"
+URI_REGEX = rf"{URL_REGEX[:-1]}(?:#?|[#?]\S+)$"
 URI = colander.Regex(URI_REGEX, msg=colander._("Must be a URI"), flags=re.IGNORECASE)
 STRING_FORMATTERS.update({
     "uri": {"converter": BaseStringTypeConverter, "validator": URI},
@@ -1025,6 +1026,9 @@ def _check_deserialize(node, cstruct):
     def _deserialize_remap(node, cstruct, var_map, var_name, has_const_child):
         invalid_var = colander.Invalid(node, value=var_map)
         try:
+            # ensure to use a copy to avoid modifying a structure passed down to here since we pop variable-mapped keys
+            cstruct = copy.deepcopy(cstruct)
+
             # Substitute real keys with matched variables to run full deserialize so
             # that mapping can find nodes name against attribute names, then re-apply originals.
             # We must do this as non-variable sub-schemas could be present, and we must also
@@ -1072,7 +1076,9 @@ def _deserialize_remap(node, cstruct, var_map, var_name, has_const_child):
                 if mapped is None and node.missing is colander.required:
                     raise colander.Invalid(node, value=cstruct)
                 for var_mapped in mapped:
-                    result[var_mapped["name"]] = var_mapped["cstruct"]
+                    # variable schema validation failed, but it is not marked as 'required'
+                    if var_mapped["cstruct"] not in [colander.drop, colander.null]:
+                        result[var_mapped["name"]] = var_mapped["cstruct"]
         except colander.Invalid as invalid:
             if invalid.msg:
                 invalid_var.msg = invalid.msg
@@ -1100,7 +1106,7 @@ def _deserialize_impl(self, cstruct):
         var_map_invalid = {}  # type: Dict[str, colander.Invalid]
         for var_child in var_children:
             var = getattr(var_child, self._variable, None)
-            var_map[var] = []
+            var_map.setdefault(var, [])
             var_msg = f"Requirement not met under variable: {var}."
             var_map_invalid[var] = colander.Invalid(node=self, msg=var_msg, value=cstruct)
             # attempt to find any sub-node matching the sub-schema under variable
@@ -1135,10 +1141,10 @@ def _deserialize_impl(self, cstruct):
                         # use position as tested child field name for later reference by invalid schema
                         var_map_invalid[var].add(invalid, pos=child_key)
 
-            var_val = var_map.get(var, colander.null)
-            if var_val is colander.null:
-                # allow unmatched variable item under mapping if it is not required
-                if var_child.missing is colander.drop:
+            var_mapped = var_map.get(var, [])
+            if not var_mapped:
+                # allow unmatched/unprovided variable item under mapping if it is not required
+                if var_child.missing in [colander.drop, colander.null]:
                     continue
                 # if required, don't waste more time doing lookup
                 # fail immediately since this variable schema is missing

weaver/wps_restapi/swagger_definitions.py

@@ -89,7 +89,6 @@
 from weaver.visibility import Visibility
 from weaver.wps_restapi.colander_extras import (
     NO_DOUBLE_SLASH_PATTERN,
-    URI,
     AllOfKeywordSchema,
     AnyOfKeywordSchema,
     BoundedRange,
@@ -387,11 +386,31 @@ class Tag(ExtendedSchemaNode):
 
 
 class URL(ExtendedSchemaNode):
+    """
+    String format that will be automatically mapped to a URL-pattern validator.
+
+    .. seealso::
+        - :data:`weaver.wps_restapi.colander_extras.URL`
+        - :class:`weaver.wps_restapi.colander_extras.ExtendedSchemaBase`
+    """
     schema_type = String
     description = "URL reference."
     format = "url"
 
 
+class URI(ExtendedSchemaNode):
+    """
+    String format that will be automatically mapped to a URI-pattern validator.
+
+    .. seealso::
+        - :data:`weaver.wps_restapi.colander_extras.URI`
+        - :class:`weaver.wps_restapi.colander_extras.ExtendedSchemaBase`
+    """
+    schema_type = String
+    description = "URI reference."
+    format = "uri"
+
+
 class Email(ExtendedSchemaNode):
     schema_type = String
     description = "Email recipient."
@@ -4990,50 +5009,63 @@ class CWLBase(ExtendedMappingSchema):
     cwlVersion = CWLVersion()
 
 
-class CWLNamespaces(ExtendedMappingSchema):
+class CWLNamespaces(StrictMappingSchema):
+    """
+    Mapping of :term:`CWL` namespace definitions for shorthand notation.
+
+    .. note::
+        Use a combination of `strict` mapping and ``variable`` (see ``var`` field) such that any additional namespace
+        other than the ones explicitly listed are allowed, but if provided, they must succeed URI validation minimally.
+        If no additional namespace is provided, including none at all, the mapping definition remains valid because
+        of ``missing=drop`` under ``var``. If a URI is invalid for additional namespaces, the failing validation causes
+        the property to be unmapped to the variable, which leads to an ``"unknown"`` property raised by the `strict`
+        mapping. For explicit URI definitions, the specific URI combinations provided must be matched exactly to
+        succeed. This ensures that no invalid mapping gets applied for commonly-known URI namespaces.
+    """
     name = "$namespaces"
     title = "CWL Namespaces Mapping"
-    cwl = URL(
+    description = "Mapping of CWL namespace definitions for shorthand notation."
+    var = URI(variable="{namespace}", missing=drop)
+    cwl = URI(
         missing=drop,
         name=CWL_NAMESPACE,
-        validator=OneOf([CWL_NAMESPACE_URL]),
+        validator=OneOf([CWL_NAMESPACE_URL, CWL_NAMESPACE_URL.rstrip("#")]),
     )
-    cwltool = URL(
+    cwltool = URI(
         missing=drop,
         name=CWL_NAMESPACE_CWLTOOL,
-        validator=OneOf([CWL_NAMESPACE_CWLTOOL_URL]),
+        validator=OneOf([CWL_NAMESPACE_CWLTOOL_URL, CWL_NAMESPACE_CWLTOOL_URL.rstrip("#")]),
     )
-    edam = URL(
+    edam = URI(
         missing=drop,
         name=EDAM_NAMESPACE,
-        validator=OneOf([EDAM_NAMESPACE_URL]),
+        validator=OneOf([EDAM_NAMESPACE_URL, EDAM_NAMESPACE_URL.rstrip("#")]),
     )
-    iana = URL(
+    iana = URI(
         missing=drop,
         name=IANA_NAMESPACE,
-        validator=OneOf([IANA_NAMESPACE_URL]),
+        validator=OneOf([IANA_NAMESPACE_URL, IANA_NAMESPACE_URL.rstrip("#")]),
     )
-    ogc = URL(
+    ogc = URI(
         missing=drop,
         name=OGC_NAMESPACE,
-        validator=OneOf([OGC_NAMESPACE_URL]),
+        validator=OneOf([OGC_NAMESPACE_URL, OGC_NAMESPACE_URL.rstrip("#")]),
     )
-    opengis = URL(
+    opengis = URI(
         missing=drop,
         name=OPENGIS_NAMESPACE,
-        validator=OneOf([OPENGIS_NAMESPACE_URL]),
+        validator=OneOf([OPENGIS_NAMESPACE_URL, OPENGIS_NAMESPACE_URL.rstrip("#")]),
     )
-    s = URL(
+    s = URI(
         missing=drop,
         name=CWL_NAMESPACE_SCHEMA,
-        validator=OneOf([CWL_NAMESPACE_SCHEMA_URL]),
+        validator=OneOf([CWL_NAMESPACE_SCHEMA_URL, CWL_NAMESPACE_SCHEMA_URL.rstrip("#")]),
     )
-    weaver = URL(
+    weaver = URI(
         missing=drop,
         name=CWL_NAMESPACE_WEAVER,
-        validator=OneOf([CWL_NAMESPACE_WEAVER_URL]),
+        validator=OneOf([CWL_NAMESPACE_WEAVER_URL, CWL_NAMESPACE_WEAVER_URL.rstrip("#")]),
     )
-    var = URL(variable="{namespace}", missing=drop)
 
 
 class CWLSchemas(ExtendedSequenceSchema):
@@ -6096,7 +6128,7 @@ class ErrorJsonResponseBodySchema(ExtendedMappingSchema):
     cause = ErrorCause(missing=drop)
     value = ErrorCause(missing=drop)
     error = ErrorDetail(missing=drop)
-    instance = ExtendedSchemaNode(String(), validator=URI, missing=drop)
+    instance = URI(missing=drop)
     exception = OWSExceptionResponse(missing=drop)