API reference

class AssertionHint(BaseModel):
    """Base model for an assertion hint attached to a CWL process.

    Attributes:
        parent_workflow: Parsed CWL process containing the hint, when available.
    """

    parent_workflow: Any | None = None

    @property
    @computed_field
    @abstractmethod
    def annotation(self) -> str:
        """Return the annotation key used to identify this hint."""
        pass

    @abstractmethod
    def validator(self) -> BaseValidator:
        """Create a validator configured from this hint."""
        pass

class BaseValidator(ABC):
    """Interface for validators that check a mapping of workflow inputs."""

    @abstractmethod
    def validate_inputs(self, data: Mapping[str, Any]) -> ProblemDetails | None:
        """Check workflow inputs against the configured assertions.

        Args:
            data: Workflow input names mapped to their values.

        Returns:
            Problem details describing violations, or None when validation succeeds.

        Implementations may raise an exception when validation cannot be performed.
        """
        pass

class Cql2FilterHint(AssertionHint):
    """Configure CQL2 predicates for workflow inputs.

    Attributes:
        custom_functions: Optional trusted Python source executed to define
            functions available to the CQL2 evaluator.
        queries: Predicates that must evaluate to a truthy value.
    """

    @staticmethod
    def get_annotation_name() -> str:
        """Return the annotation key for CQL2 input filters."""
        return "eoap.ogc.org/inputs-cql2-filter"

    custom_functions: str | None = None
    queries: list[Cql2Query]

    @property
    def annotation(self) -> str:
        """Return the annotation key for CQL2 input filters."""
        return Cql2FilterHint.get_annotation_name()

    @model_serializer
    def ser_model(self) -> Mapping[str, Any]:
        """Serialize the queries as an annotation payload, omitting custom functions."""
        return {"queries": self.queries}

    def validator(self) -> BaseValidator:
        """Create a CQL2 validator with the configured queries and custom functions."""
        from .cql2_validator import Cql2Validator

        return Cql2Validator(custom_functions=self.custom_functions, queries=self.queries)

class Cql2Query(BaseModel):
    """Describe a CQL2 assertion and the violation reported when it is false.

    Attributes:
        id: Rule identifier used as the error pointer.
        cql2: CQL2 Text expression or CQL2 JSON expression mapping.
        message: Violation message to report when the predicate is false.
    """

    id: str
    cql2: str | Mapping[str, Any]
    message: str

class JSONSchemaHint(AssertionHint):
    """Configure JSON Schema validation of workflow inputs.

    Attributes:
        json_schema: Explicit schema to apply, including an explicitly empty schema.
            When omitted, derive a schema from parent_workflow if available;
            otherwise use an empty schema.
    """

    json_schema: Mapping[str, Any] = {}

    @staticmethod
    def get_annotation_name() -> str:
        """Return the annotation key for JSON Schema input constraints."""
        return "eoap.ogc.org/inputs-json-schema"

    @property
    def annotation(self) -> str:
        """Return the annotation key for JSON Schema input constraints."""
        return JSONSchemaHint.get_annotation_name()

    def _get_schema(self) -> Mapping[str, Any]:
        """Return the explicit schema, or derive it from the parent CWL process."""
        if "json_schema" in self.model_fields_set or self.parent_workflow is None:
            return self.json_schema
        schema = BaseCWLtypes2OGCConverter(self.parent_workflow).get_inputs_json_schema()
        if not isinstance(schema, Mapping):
            raise TypeError("The CWL converter must return a JSON Schema mapping")
        return schema

    @model_serializer
    def ser_model(self) -> Mapping[str, Any]:
        """Serialize the hint as its effective JSON Schema document."""
        return self._get_schema()

    def validator(self) -> BaseValidator:
        """Create a JSON Schema validator using the effective schema."""
        from .jsonschema_validator import JSONSchemaValidator

        return JSONSchemaValidator(schema=self._get_schema())

class RegoPolicyHint(AssertionHint):
    """Configure Rego policy queries for workflow inputs.

    Attributes:
        module: Rego source code defining the policy.
        queries: Queries whose returned expressions describe violations.
    """

    @staticmethod
    def get_annotation_name() -> str:
        """Return the annotation key for Rego input policies."""
        return "eoap.ogc.org/inputs-rego-policy"

    module: str
    queries: list[str]

    @property
    def annotation(self) -> str:
        """Return the annotation key for Rego input policies."""
        return RegoPolicyHint.get_annotation_name()

    @model_serializer
    def ser_model(self) -> Mapping[str, Any]:
        """Serialize the policy module and queries as an annotation payload."""
        return {"queries": self.queries, "module": self.module}

    def validator(self) -> BaseValidator:
        """Create a Rego validator using this policy module and its queries."""
        from .rego_validator import RegoValidator

        return RegoValidator(queries=self.queries, module=self.module)

class JSONSchemaRegistry:
    """Retrieve JSON or YAML schema documents through a Requests session."""

    def __init__(
        self,
        timeout: int = _REMOTE_SCHEMA_TIMEOUT_SECONDS,
    ):
        """Create a session for schema retrieval.

        Args:
            timeout: Timeout in seconds passed to each schema HTTP request.
        """
        self.session = requests.Session()
        self.session.headers.update({"User-Agent": "assertions-mate"})
        self.timeout = timeout

    @staticmethod
    def _schema_format_from_content_type(content_type: str | None) -> str | None:
        """Infer json or yaml from a media type, or return None if unrecognized."""
        if not content_type:
            return None

        media_type = content_type.split(";", 1)[0].strip().lower()

        if media_type == "application/json" or media_type.endswith("+json"):
            return "json"

        if media_type in {
            "application/yaml",
            "application/x-yaml",
            "text/yaml",
            "text/x-yaml",
        } or media_type.endswith("+yaml"):
            return "yaml"

        return None

    @staticmethod
    def _schema_format_from_uri(uri: str) -> str | None:
        """Infer json or yaml from a URI path suffix, or return None."""
        path = urlparse(uri).path.lower()

        if path.endswith(".json"):
            return "json"

        if path.endswith((".yaml", ".yml")):
            return "yaml"

        return None

    def _parse_schema(
        self,
        payload: bytes,
        uri: str,
        content_type: str | None = None,
    ) -> Any:
        """Decode a UTF-8 schema document as JSON or YAML.

        Args:
            payload: Schema document bytes.
            uri: Document URI, used for format detection and error reporting.
            content_type: Optional HTTP media type, preferred over the URI suffix.

        Returns:
            The parsed document. If the preferred parser fails, try the other format.

        Raises:
            UnicodeDecodeError: If the payload is not valid UTF-8.
            ValueError: If both JSON and YAML parsing fail.
        """
        text = payload.decode("utf-8")
        preferred_format = (
            self._schema_format_from_content_type(content_type)
            or self._schema_format_from_uri(uri)
            or "json"
        )
        parsers: tuple[Callable[[str], object], ...] = (json.loads, YAML().load)
        if preferred_format == "yaml":
            parsers = parsers[::-1]
        preferred_parser, fallback_parser = parsers
        try:
            return preferred_parser(text)
        except (ValueError, YAMLError):
            pass
        try:
            return fallback_parser(text)
        except (ValueError, YAMLError) as error:
            raise ValueError(f"Unable to parse JSON Schema from {uri}") from error

    @staticmethod
    def _schema_request_uri(uri: str) -> str:
        """Preserve a URI scheme or convert an existing path to a file URI.

        Raises:
            NoSuchResource: If a schemeless path does not exist.

        Converting a path does not install a Requests adapter for file URIs.
        """
        parsed_uri = urlparse(uri)
        if parsed_uri.scheme:
            return uri

        path = Path(uri)
        if not path.exists():
            raise NoSuchResource(uri)

        return path.absolute().as_uri()

    def load_schema(self, uri: str) -> Any:
        """Fetch a schema document and parse it as JSON or YAML.

        Args:
            uri: Schema location supported by the configured Requests session.
                The default session supports HTTP and HTTPS, not local files.

        Returns:
            The parsed schema document.

        Raises:
            NoSuchResource: If a path is missing, the URI scheme has no adapter,
                or the server returns HTTP 404.
            requests.exceptions.RequestException: If the request otherwise fails.
            UnicodeDecodeError: If the response is not valid UTF-8.
            ValueError: If the response cannot be parsed as JSON or YAML.
        """
        try:
            response = self.session.get(
                self._schema_request_uri(uri),
                timeout=self.timeout,
            )
        except requests.exceptions.InvalidSchema as error:
            raise NoSuchResource(uri) from error

        if response.status_code == HTTPStatus.NOT_FOUND.value:
            raise NoSuchResource(uri)

        response.raise_for_status()

        return self._parse_schema(
            payload=response.content,
            uri=uri,
            content_type=response.headers.get("Content-Type"),
        )

    def retrieve(self, uri: str) -> Resource[Any]:
        """Load a URI as a Draft 2020-12 resource for reference resolution.

        Args:
            uri: Schema location to fetch through load_schema.

        Returns:
            A schema resource wrapping the parsed document.

        Retrieval and parsing errors from load_schema propagate to the caller.
        """
        return DRAFT202012.create_resource(self.load_schema(uri))

    def as_referencing_registry(self) -> ReferencingRegistry[Any]:
        # mypy does not understand the attrs field alias used by referencing.
        """Create a referencing registry backed by this instance's retrieval callback."""
        return ReferencingRegistry(retrieve=self.retrieve)  # type: ignore[call-arg]

class JSONSchemaValidator(BaseValidator):
    """Validate input mappings against a JSON Schema Draft 2020-12 schema."""

    def __init__(
        self,
        schema: Mapping[str, Any],
    ):
        """Configure schema validation and retrieval of referenced schemas.

        Args:
            schema: JSON Schema document describing accepted workflow inputs.
        """
        self.validator = Draft202012Validator(
            schema,
            registry=JSONSchemaRegistry().as_referencing_registry(),
        )

    def validate_inputs(self, data: Mapping[str, Any]) -> ProblemDetails | None:
        """Collect schema violations for a workflow input mapping.

        Args:
            data: Workflow input names mapped to their values.

        Returns:
            InvalidBodyPropertyFormat containing schema violations, or None when
            the inputs conform to the schema.

        Schema evaluation and reference resolution errors propagate to the caller.
        """
        errors_list = []

        for error in self.validator.iter_errors(data):
            print(error.__dict__)

            errors_list.append(
                ErrorDetail(
                    detail=error.message,
                    pointer=f"#/{'/'.join(error.path)}"
                    if error.path and isinstance(error.path, str)
                    else None,
                    code=".".join(error.schema_path),
                )
            )

        if errors_list:
            return InvalidBodyPropertyFormat(errors=errors_list)

        return None

class RegoValidator(BaseValidator):
    """Run Rego queries and collect their returned expressions as violations."""

    def __init__(self, module: str, queries: list[str]) -> None:
        """Load a policy module into a Rego interpreter.

        Args:
            module: Rego source registered under the interpreter module name workflow.
            queries: Queries to run after setting the workflow inputs.

        Policy loading errors propagate from the interpreter.
        """
        self.rego = Interpreter()
        self.rego.add_module("workflow", module)
        self.queries = queries

    def validate_inputs(self, data: Mapping[str, Any]) -> ProblemDetails | None:
        """Set the Rego input document and collect query result expressions.

        Args:
            data: Workflow input mapping exposed to policies as input.

        Returns:
            BusinessRuleViolation containing one error per returned expression,
            with the query as its pointer, or None when no expressions are returned.
            Expressions are converted to strings without testing their truthiness.

        Interpreter errors propagate to the caller.
        """
        errors_list: list[ErrorDetail] = []

        self.rego.set_input(Input(data))

        for query in self.queries:
            for result in self.rego.query(query):
                # result.expressions is a list of Expression objects
                exprs = result.expressions
                if not exprs:  # safety check
                    continue

                errors_list.extend(ErrorDetail(pointer=query, detail=str(expr)) for expr in exprs)

        if errors_list:
            return BusinessRuleViolation(errors=errors_list)

        return None

class Cql2EvaluationError(RuntimeError):
    """A rule could not be evaluated, rather than evaluating to false."""

class Cql2Validator(BaseValidator):
    """Validate workflow inputs with CQL2 predicates and optional Python functions."""

    def __init__(self, queries: list[Cql2Query], custom_functions: str | None = None) -> None:
        """Initialize the native evaluator and retain the configured queries.

        Args:
            queries: Rules to evaluate against each input mapping.
            custom_functions: Trusted Python source executed immediately to populate
                the evaluator function map. Exceptions from this code propagate.
        """
        function_map: dict[str, Any] = {}

        if custom_functions:
            # SECURITY: CWL custom functions are trusted application code.
            # Loading untrusted workflows would permit arbitrary code execution.
            exec(custom_functions, function_map)  # nosec B102

        self.evaluator = NativeEvaluator(function_map=function_map, use_getattr=False)

        self.queries = queries

    def validate_inputs(self, data: Mapping[str, Any]) -> ProblemDetails | None:
        """Evaluate every configured CQL2 rule against the inputs.

        Args:
            data: Workflow input names mapped to their values.

        Returns:
            BusinessRuleViolation for malformed rules or false predicates, or None
            when all rules pass. Error pointers identify the corresponding rules.

        Raises:
            Cql2EvaluationError: If a parsed rule cannot be compiled or evaluated.
                Processing stops at that rule rather than returning a violation.
        """
        errors_list = []

        for query in self.queries:
            errors_list.extend(self._validate_query(query, data))

        if errors_list:
            return BusinessRuleViolation(errors=errors_list)

        return None

    def _validate_query(self, query: Cql2Query, data: Mapping[str, Any]) -> list[ErrorDetail]:
        """Parse and evaluate one rule, returning its validation errors.

        Raises:
            Cql2EvaluationError: If compilation or predicate execution fails.
        """
        errors_list = []
        ast = None

        if isinstance(query.cql2, str):
            try:
                ast = parse_cql2_text(query.cql2)
            except Exception as error:
                errors_list.append(
                    ErrorDetail(
                        pointer=query.id,
                        detail=f"Filter does not look like a valid CQL2 Text encoded sentece: {error}",
                    )
                )
        elif isinstance(query.cql2, dict):
            try:
                ast = parse_cql2_json(_to_builtin(query.cql2))
            except Exception as error:
                errors_list.append(
                    ErrorDetail(
                        pointer=query.id,
                        detail=f"Filter does not look like a valid CQL2 JSON encoded structure: {error}",
                    )
                )
        else:
            errors_list.append(
                ErrorDetail(
                    pointer=query.id,
                    detail=f"Filter is expressed in an unrecognizible format: {type(query.cql2)}",
                )
            )

        if ast and not self._evaluate_predicate(ast, data, query.id):
            errors_list.append(ErrorDetail(pointer=query.id, detail=query.message))

        return errors_list

    def _evaluate_predicate(self, ast: object, data: Mapping[str, Any], rule_id: str) -> bool:
        """Evaluate a parsed rule, wrapping evaluator failures with its identifier."""
        try:
            predicate = self.evaluator.evaluate(ast)
            return bool(predicate(data))
        except Exception as error:
            logger.opt(exception=True).debug("Failed to evaluate CQL2 rule '{}'", rule_id)
            raise Cql2EvaluationError(f"Could not evaluate rule '{rule_id}'") from error