feat(jans-pycloudlib): secure mounted configuration schema

jans-pycloudlib/jans/pycloudlib/config/file_config.py

@@ -12,8 +12,9 @@
 class FileConfig(BaseConfig):
     def __init__(self) -> None:
         filepath = os.environ.get("CN_CONFIGURATOR_CONFIGURATION_FILE", "/etc/jans/conf/configuration.json")
+        key_file = os.environ.get("CN_CONFIGURATOR_KEY_FILE", "/etc/jans/conf/configuration.key")
 
-        out, err, code = load_schema_from_file(filepath, exclude_secret=True)
+        out, err, code = load_schema_from_file(filepath, exclude_secret=True, key_file=key_file)
         if code != 0:
             logger.warning(f"Unable to load configmaps from file {filepath}; error={err}; local configmaps will be excluded")
 

jans-pycloudlib/jans/pycloudlib/schema/__init__.py

@@ -4,6 +4,7 @@
 import logging
 import re
 from base64 import b64decode
+from contextlib import suppress
 
 import pem
 from fqdn import FQDN
@@ -20,6 +21,7 @@
 from marshmallow.validate import OneOf
 from marshmallow.validate import Predicate
 from marshmallow.validate import Range
+from sprig_aes import sprig_decrypt_aes
 
 logger = logging.getLogger(__name__)
 
@@ -884,38 +886,72 @@ class Meta:
     _configmap = Nested(ConfigmapSchema, required=True)
 
 
-def load_schema_from_file(path, exclude_configmap=False, exclude_secret=False):
+def load_schema_from_file(path, exclude_configmap=False, exclude_secret=False, key_file=""):
     """Loads schema from file."""
-    out = {}
-    err = {}
-    code = 0
+    out, err, code = maybe_encrypted_schema(path, key_file)
 
-    try:
-        with open(path) as f:
-            docs = json.loads(f.read())
-    except (IOError, ValueError) as exc:
-        err = exc
-        code = 1
+    if code != 0:
         return out, err, code
 
     # dont exclude attributes
-    exclude_attrs = False
+    exclude_attrs = []
 
     # exclude configmap from loading mechanism
     if exclude_configmap:
         key = "_configmap"
         exclude_attrs = [key]
-        docs.pop(key, None)
+        out.pop(key, None)
 
     # exclude secret from loading mechanism
     if exclude_secret:
         key = "_secret"
         exclude_attrs = [key]
-        docs.pop(key, None)
+        out.pop(key, None)
 
     try:
-        out = ConfigurationSchema().load(docs, partial=exclude_attrs)
+        out = ConfigurationSchema().load(out, partial=exclude_attrs)
     except ValidationError as exc:
         err = exc.messages
         code = 1
     return out, err, code
+
+
+def load_schema_key(path):
+    try:
+        with open(path) as f:
+            key = f.read().strip()
+    except FileNotFoundError:
+        key = ""
+    return key
+
+
+def maybe_encrypted_schema(path, key_file):
+    out, err, code = {}, {}, 0
+
+    try:
+        # read schema as raw string
+        with open(path) as f:
+            raw_txt = f.read()
+    except FileNotFoundError as exc:
+        err = {
+            "error": f"Unable to load schema {path}",
+            "reason": exc,
+        }
+        code = exc.errno
+    else:
+        if key := load_schema_key(key_file):
+            # try to decrypt schema (if applicable)
+            with suppress(ValueError):
+                raw_txt = sprig_decrypt_aes(raw_txt, key)
+
+        try:
+            out = json.loads(raw_txt)
+        except (json.decoder.JSONDecodeError, UnicodeDecodeError) as exc:
+            err = {
+                "error": f"Unable to decode JSON from {path}",
+                "reason": exc,
+            }
+            code = 1
+
+    # finalized results
+    return out, err, code

jans-pycloudlib/jans/pycloudlib/secret/file_secret.py

@@ -12,8 +12,9 @@
 class FileSecret(BaseSecret):
     def __init__(self) -> None:
         filepath = os.environ.get("CN_CONFIGURATOR_CONFIGURATION_FILE", "/etc/jans/conf/configuration.json")
+        key_file = os.environ.get("CN_CONFIGURATOR_KEY_FILE", "/etc/jans/conf/configuration.key")
 
-        out, err, code = load_schema_from_file(filepath, exclude_configmap=True)
+        out, err, code = load_schema_from_file(filepath, exclude_configmap=True, key_file=key_file)
         if code != 0:
             logger.warning(f"Unable to load secrets from file {filepath}; error={err}; local secrets will be excluded")
 

jans-pycloudlib/tests/test_schema.py

@@ -159,3 +159,51 @@ def test_random_optional_scopes(value):
 
     with pytest.raises(ValidationError):
         ConfigmapSchema().validate_optional_scopes(value)
+
+
+def test_load_schema_key(tmpdir):
+    from jans.pycloudlib.schema import load_schema_key
+
+    src = tmpdir.join("configuration.key")
+    src.write("abcd")
+    assert load_schema_key(str(src)) == "abcd"
+
+
+def test_maybe_encrypted_schema_file_missing():
+    from jans.pycloudlib.schema import maybe_encrypted_schema
+
+    _, err, _ = maybe_encrypted_schema("/path/to/schema/file", "/path/to/schema/key")
+    assert "error" in err
+
+
+def test_maybe_encrypted_schema(tmpdir):
+    from jans.pycloudlib.schema import maybe_encrypted_schema
+
+    src = tmpdir.join("configuration.json")
+    src.write("zLBGM41dAfA2JuIkVHRKa+/WwVo/8oQAdD0LUT3jGfhqp/euYdDhf+kTiKwfb1Sv28zYL12JlO+3oSl6ZlhiTw==")
+
+    src_key = tmpdir.join("configuration.key")
+    src_key.write("6Jsv61H7fbkeIkRvUpnZ98fu")
+
+    out, _, _ = maybe_encrypted_schema(str(src), str(src_key))
+    assert out == {"_configmap": {"hostname": "example.com"}}
+
+
+def test_schema_exclude_configmap(tmpdir):
+    from jans.pycloudlib.schema import load_schema_from_file
+
+    src = tmpdir.join("configuration.json")
+    src.write('{"_configmap": {}, "_secret": {"admin_password": "Test1234#"}}')
+
+    out, _, code = load_schema_from_file(str(src), exclude_configmap=True)
+    assert "_configmap" not in out and code == 0
+
+
+def test_schema_exclude_secret(tmpdir):
+    from jans.pycloudlib.schema import load_schema_from_file
+
+    src = tmpdir.join("configuration.json")
+    src.write('{"_configmap": {"city": "Austin", "country_code": "US", "admin_email": "[email protected]", "hostname": "example.com", "orgName": "Example Inc.", "state": "TX"}, "_secret": {}}')
+
+    out, _, code = load_schema_from_file(str(src), exclude_secret=True)
+    assert "_secret" not in out and code == 0