Enh/secrets json encoders

docs/api_reference/models/secretconfig.md

@@ -0,0 +1,5 @@
+# secretconfig
+
+::: oteapi.models.secretconfig
+    options:
+      show_if_no_docstring: true

docs/api_reference/settings.md

@@ -0,0 +1,3 @@
+# settings
+
+::: oteapi.settings

oteapi/models/__init__.py

@@ -10,6 +10,7 @@
 from .genericconfig import AttrDict, GenericConfig
 from .mappingconfig import MappingConfig, RDFTriple
 from .resourceconfig import ResourceConfig
+from .secretconfig import SecretConfig
 from .sessionupdate import SessionUpdate
 from .transformationconfig import TransformationConfig, TransformationStatus
 from .triplestoreconfig import TripleStoreConfig
@@ -28,6 +29,7 @@
     "TransformationConfig",
     "TransformationStatus",
     "TripleStoreConfig",
+    "SecretConfig",
 )
 
 StrategyConfig = Union[

oteapi/models/functionconfig.py

@@ -1,20 +1,14 @@
 """Pydantic Function Configuration Data Model."""
-from typing import Optional
-
-from pydantic import Field, SecretStr
+from pydantic import Field
 
 from oteapi.models.genericconfig import GenericConfig
+from oteapi.models.secretconfig import SecretConfig
 
 
-class FunctionConfig(GenericConfig):
+class FunctionConfig(GenericConfig, SecretConfig):
     """Function Strategy Data Configuration."""
 
     functionType: str = Field(
         ...,
         description=("Type of registered function strategy."),
     )
-
-    secret: Optional[SecretStr] = Field(
-        None,
-        description="Authorization secret given when executing a function.",
-    )

oteapi/models/resourceconfig.py

@@ -1,9 +1,10 @@
 """Pydantic Resource Configuration Data Model."""
 from typing import TYPE_CHECKING, Optional
 
-from pydantic import AnyUrl, Field, SecretStr, root_validator
+from pydantic import AnyUrl, Field, root_validator
 
 from oteapi.models.genericconfig import GenericConfig
+from oteapi.models.secretconfig import SecretConfig
 
 if TYPE_CHECKING:  # pragma: no cover
     from typing import Any, Dict
@@ -15,7 +16,7 @@ class HostlessAnyUrl(AnyUrl):
     host_required = False
 
 
-class ResourceConfig(GenericConfig):
+class ResourceConfig(GenericConfig, SecretConfig):
     """Resource Strategy Data Configuration.
 
     Important:
@@ -76,10 +77,6 @@ class ResourceConfig(GenericConfig):
         None,
         description="The entity responsible for making the resource/item available.",
     )
-    secret: Optional[SecretStr] = Field(
-        None,
-        description="Authorization secret given when accessing a resource.",
-    )
 
     @root_validator
     def ensure_unique_url_pairs(cls, values: "Dict[str, Any]") -> "Dict[str, Any]":

oteapi/models/secretconfig.py

@@ -0,0 +1,57 @@
+"""AttrDict for specifying user credentials or secrets."""
+import json
+from typing import TYPE_CHECKING, Optional
+
+from pydantic import BaseModel, Field, SecretStr
+
+from oteapi.settings import settings
+
+if TYPE_CHECKING:  # pragma: no cover
+    from typing import Any, Callable
+
+
+def json_dumps(model: dict, default: "Callable[[Any], Any]") -> "str":
+    """Alternative function for dumping exposed
+    secrets to json when model is serialized.
+
+    Parameters:
+        model: The pydantic model to serialize.
+        default: A pass-through to the standard `json.dumps()`'s `default` parameter.
+            From the `json.dumps()` doc-string: `default(obj)` is a function that should
+            return a serializable version of `obj` or raise `TypeError`.
+            The default simply raises `TypeError`.
+
+    Returns:
+        The result of `json.dumps()` after handling possible secrets.
+
+    """
+    return json.dumps(
+        {
+            key: (
+                value.get_secret_value()
+                if settings.expose_secrets and isinstance(value, SecretStr)
+                else value
+            )
+            for key, value in model.items()
+        },
+        default=default,
+    )
+
+
+class SecretConfig(BaseModel, json_dumps=json_dumps):
+    """Simple model for handling secret in other config-models."""
+
+    user: Optional[SecretStr] = Field(None, description="User name for authentication.")
+    password: Optional[SecretStr] = Field(
+        None, description="Password for authentication."
+    )
+    token: Optional[SecretStr] = Field(
+        None,
+        description="An access token for providing access and meta data to an application.",
+    )
+    client_id: Optional[SecretStr] = Field(
+        None, description="Client ID for an OAUTH2 client."
+    )
+    client_secret: Optional[SecretStr] = Field(
+        None, description="Client secret for an OAUTH2 client."
+    )

oteapi/models/transformationconfig.py

@@ -8,9 +8,10 @@
 from enum import Enum
 from typing import List, Optional
 
-from pydantic import BaseModel, Field, SecretStr
+from pydantic import BaseModel, Field
 
 from oteapi.models.genericconfig import GenericConfig
+from oteapi.models.secretconfig import SecretConfig
 
 
 class ProcessPriority(str, Enum):
@@ -29,7 +30,7 @@ class ProcessPriority(str, Enum):
     HIGH = "High"
 
 
-class TransformationConfig(GenericConfig):
+class TransformationConfig(GenericConfig, SecretConfig):
     """Transformation Strategy Data Configuration."""
 
     transformationType: str = Field(
@@ -52,10 +53,6 @@ class TransformationConfig(GenericConfig):
         ProcessPriority.MEDIUM,
         description="Define the process priority of the transformation execution.",
     )
-    secret: Optional[SecretStr] = Field(
-        None,
-        description="Authorization secret given when running a transformation.",
-    )
 
 
 class TransformationStatus(BaseModel):

oteapi/models/triplestoreconfig.py

@@ -1,10 +1,16 @@
 """Pydantic TripleStore Configuration Data Model."""
-from pydantic import Field, SecretStr
+from typing import TYPE_CHECKING
 
-from oteapi.models.genericconfig import AttrDict
+from pydantic import Field, root_validator
 
+from oteapi.models.genericconfig import GenericConfig
+from oteapi.models.secretconfig import SecretConfig
 
-class TripleStoreConfig(AttrDict):
+if TYPE_CHECKING:  # pragma: no cover
+    from typing import Any, Dict
+
+
+class TripleStoreConfig(GenericConfig, SecretConfig):
     """TripleStore Configuration.
 
     This is a configuration for the
@@ -26,11 +32,21 @@ class TripleStoreConfig(AttrDict):
         ...,
         description="AllegroGraph port number.",
     )
-    agraphUser: str = Field(
-        ...,
-        description="AllegroGraph user name.",
-    )
-    agraphPassword: SecretStr = Field(
-        ...,
-        description="AllegroGraph user password.",
-    )
+
+    @root_validator
+    def ensure_user_pass(cls, values: "Dict[str, Any]") -> "Dict[str, Any]":
+        """Ensure that user/password are set, since they are optional in the SecretConfig."""
+        if not all(values.get(_) for _ in ["user", "password"]):
+            raise ValueError("User and password must be defined.")
+        return values
+
+    class Config:
+        """Pydantic configuration for TripleStoreConfig."""
+
+        fields = {
+            "token": {"exclude": True},
+            "client_id": {"exclude": True},
+            "client_secret": {"exclude": True},
+        }
+        """The `fields`-config enables that `token`, `client_id` and `client_secret`
+        will be excluded, when the model is serialized."""

oteapi/settings.py

@@ -0,0 +1,31 @@
+"""BaseSettings for oteapi-core.
+This `configuration/settings`-class is intended to be incorporated as a
+parentclass into the configuration of an FastAPI application.
+See `https://fastapi.tiangolo.com/advanced/settings/` as reference.
+
+Otherwise, check `https://github.com/EMMC-ASBL/oteapi-services/blob/master/app/main.py`
+for a direct example of an inclusion of the OTE api and its settings into an FastAPI instance.
+"""
+from pydantic import BaseSettings, Field
+
+
+class OteApiCoreSettings(BaseSettings):
+    """Basic configuration for the oteapi-core."""
+
+    expose_secrets: bool = Field(
+        False,
+        description="Whether `SecretStr` in `pydantic` should be exposed or not.\n\n"
+        "!!! warning\n"
+        "    Depending on the configuration and user management of the services"
+        "    using oteapi-core, secrets might be readable by other users when serialized!"
+        "    This especially takes place when then models and configs are put into the cache."
+        "    Hence be careful while using this option in production.",
+    )
+
+    class Config:
+        """Pydantic config for the OteApiCoreSettings."""
+
+        env_prefix = "OTEAPI_"
+
+
+settings = OteApiCoreSettings()

oteapi/triplestore/triplestore.py

@@ -58,8 +58,8 @@ def __init__(
         self.server = AllegroGraphServer(
             self.config.agraphHost,
             self.config.agraphPort,
-            self.config.agraphUser,
-            self.config.agraphPassword.get_secret_value(),
+            self.config.user.get_secret_value(),  # type: ignore [union-attr]
+            self.config.password.get_secret_value(),  # type: ignore [union-attr]
         )
 
     def add(self, triples: RDFTriple) -> None:
@@ -74,8 +74,8 @@ def add(self, triples: RDFTriple) -> None:
             self.config.repositoryName,
             host=self.config.agraphHost,
             port=self.config.agraphPort,
-            user=self.config.agraphUser,
-            password=self.config.agraphPassword.get_secret_value(),
+            user=self.config.user.get_secret_value(),  # type: ignore [union-attr]
+            password=self.config.password.get_secret_value(),  # type: ignore [union-attr]
         ) as connection:
             connection.addData(triples)
             connection.close()
@@ -127,8 +127,8 @@ def update_delete(self, sparql_query: str) -> None:
             self.config.repositoryName,
             host=self.config.agraphHost,
             port=self.config.agraphPort,
-            user=self.config.agraphUser,
-            password=self.config.agraphPassword.get_secret_value(),
+            user=self.config.user.get_secret_value(),  # type: ignore [union-attr]
+            password=self.config.password.get_secret_value(),  # type: ignore [union-attr]
         ) as connection:
             update_query = connection.prepareUpdate(query=sparql_query)
             update_query.evaluate()

tests/models/test_functionconfig.py

@@ -0,0 +1,38 @@
+"""Tests for `oteapi.models.functionconfig`"""
+
+
+def test_functionconfig() -> None:
+    """Pytest for FunctionConfig, mainly for testing the included secrets."""
+    import json
+
+    from oteapi.models.functionconfig import FunctionConfig
+    from oteapi.settings import settings
+
+    base_config = {"functionType": "foo/bar", "token": "abc"}
+    config_exposed = {
+        "user": None,
+        "password": None,
+        "token": "abc",
+        "client_id": None,
+        "client_secret": None,
+        "configuration": {},
+        "description": "Function Strategy Data Configuration.",
+        "functionType": "foo/bar",
+    }
+
+    config_hidden = {
+        "user": None,
+        "password": None,
+        "token": "**********",
+        "client_id": None,
+        "client_secret": None,
+        "configuration": {},
+        "description": "Function Strategy Data Configuration.",
+        "functionType": "foo/bar",
+    }
+
+    settings.expose_secrets = False
+    assert FunctionConfig(**base_config).json() == json.dumps(config_hidden)
+
+    settings.expose_secrets = True
+    assert FunctionConfig(**base_config).json() == json.dumps(config_exposed)

tests/models/test_secretconfig.py

@@ -0,0 +1,32 @@
+"""Tests for `oteapi.models.secretconfig`"""
+
+
+def test_secretconfig():
+    """Pytest for SecretConfig."""
+    import json
+
+    from oteapi.models.secretconfig import SecretConfig
+    from oteapi.settings import settings
+
+    base_config = {"token": "abc"}
+    config_exposed = {
+        "user": None,
+        "password": None,
+        "token": "abc",
+        "client_id": None,
+        "client_secret": None,
+    }
+
+    config_hidden = {
+        "user": None,
+        "password": None,
+        "token": "**********",
+        "client_id": None,
+        "client_secret": None,
+    }
+
+    settings.expose_secrets = False
+    assert SecretConfig(**base_config).json() == json.dumps(config_hidden)
+
+    settings.expose_secrets = True
+    assert SecretConfig(**base_config).json() == json.dumps(config_exposed)