refactor(jans-pycloudlib): rename CN_LOCK env vars to avoid conflict …

…with jans-lock (#7265)

* feat(jans-pycloudlib): add support for encoded password files

Signed-off-by: iromli <isman.firmansyah@gmail.com>

* refactor(jans-pycloudlib): rename CN_LOCK env vars to avoid conflict with jans-lock

Signed-off-by: iromli <isman.firmansyah@gmail.com>

* chore(jans-pycloudlib: remove code smell on testcase

Signed-off-by: iromli <isman.firmansyah@gmail.com>

---------

Signed-off-by: iromli <isman.firmansyah@gmail.com>
Former-commit-id: 8b5a9ab

jans-pycloudlib/jans/pycloudlib/lock/__init__.py

@@ -89,7 +89,7 @@ def _on_connection_giveup(details: Details) -> None:
 class LockManager:
     @property
     def lock_enabled(self):
-        return as_boolean(os.environ.get("CN_ENABLE_LOCK", "true"))
+        return as_boolean(os.environ.get("CN_OCI_LOCK_ENABLED", "true"))
 
     @backoff.on_exception(
         backoff.constant,
@@ -239,16 +239,16 @@ def adapter(self) -> LockAdapter:  # noqa: D412
             An instance of lock adapter class.
 
         Raises:
-            ValueError: If the value of `CN_LOCK_ADAPTER` or `CN_PERSISTENCE_TYPE` environment variable is not supported.
+            ValueError: If the value of `CN_OCI_LOCK_ADAPTER` or `CN_PERSISTENCE_TYPE` environment variable is not supported.
 
         Examples:
 
         ```py
-        os.environ["CN_LOCK_ADAPTER"] = "sql"
+        os.environ["CN_OCI_LOCK_ADAPTER"] = "sql"
         LockStorage().adapter  # returns an instance of adapter class
         ```
 
-        The adapter name is pre-populated from `CN_LOCK_ADAPTER` environment variable.
+        The adapter name is pre-populated from `CN_OCI_LOCK_ADAPTER` environment variable.
 
         Supported lock adapter name:
 
@@ -257,7 +257,7 @@ def adapter(self) -> LockAdapter:  # noqa: D412
         - `couchbase`: returns and instance of [CouchbaseLock][jans.pycloudlib.lock.couchbase_lock.CouchbaseLock]
         - `ldap`: returns and instance of [LdapLock][jans.pycloudlib.lock.ldap_lock.LdapLock]
         """
-        _adapter = os.environ.get("CN_LOCK_ADAPTER") or PersistenceMapper().mapping["default"]
+        _adapter = os.environ.get("CN_OCI_LOCK_ADAPTER") or PersistenceMapper().mapping["default"]
 
         if _adapter == "sql":
             return SqlLock()

jans-pycloudlib/jans/pycloudlib/lock/couchbase_lock.py

@@ -9,6 +9,7 @@
 
 from jans.pycloudlib.lock.base_lock import BaseLock
 from jans.pycloudlib.utils import as_boolean
+from jans.pycloudlib.utils import get_password_from_file
 
 logger = logging.getLogger(__name__)
 
@@ -24,40 +25,6 @@ def _handle_failed_request(resp) -> None:
             resp.raise_for_status()
 
 
-def _resolve_auth():
-    # list of possible password files
-    password_files = [
-        os.environ.get("CN_LOCK_PASSWORD_FILE", "/etc/jans/conf/lock_password")
-    ]
-
-    # check which user is accessing couchbase
-    user = os.environ.get("CN_COUCHBASE_SUPERUSER", "")
-
-    if user:
-        password_files.append(
-            os.environ.get("CN_COUCHBASE_SUPERUSER_PASSWORD_FILE", "/etc/jans/conf/couchbase_superuser_password")
-        )
-    else:
-        user = os.environ.get("CN_COUCHBASE_USER", "admin")
-        password_files.append(
-            os.environ.get("CN_COUCHBASE_PASSWORD_FILE", "/etc/jans/conf/couchbase_password")
-        )
-
-    # password of the running user
-    password = ""  # nosec: B105
-
-    for password_file in password_files:
-        if not os.path.isfile(password_file):
-            continue
-
-        with open(password_file) as f:
-            password = f.read().strip()
-            break
-
-    # auth credentials
-    return user, password
-
-
 class CouchbaseLock(BaseLock):
     def __init__(self):
         self.bucket_exists = False
@@ -93,7 +60,7 @@ def session(self):
 
         sess = requests.Session()
         sess.verify = False
-        sess.auth = _resolve_auth()
+        sess.auth = self._resolve_auth()
 
         # return the cached session
         return sess
@@ -217,3 +184,35 @@ def _prepare_bucket(self):
                 logger.warning(f"Unable to create required bucket {self.bucket}; status_code={resp.status_code}")
             case _:
                 resp.raise_for_status()
+
+    def _resolve_auth(self):
+        # list of possible password files
+        password_files = [
+            os.environ.get("CN_OCI_LOCK_PASSWORD_FILE", "/etc/jans/conf/oci_lock_password")
+        ]
+
+        # check which user is accessing couchbase
+        user = os.environ.get("CN_COUCHBASE_SUPERUSER", "")
+
+        if user:
+            password_files.append(
+                os.environ.get("CN_COUCHBASE_SUPERUSER_PASSWORD_FILE", "/etc/jans/conf/couchbase_superuser_password")
+            )
+        else:
+            user = os.environ.get("CN_COUCHBASE_USER", "admin")
+            password_files.append(
+                os.environ.get("CN_COUCHBASE_PASSWORD_FILE", "/etc/jans/conf/couchbase_password")
+            )
+
+        # password of the running user
+        password = ""  # nosec: B105
+
+        for password_file in password_files:
+            if not os.path.isfile(password_file):
+                continue
+
+            password = get_password_from_file(password_file)
+            break
+
+        # auth credentials
+        return user, password

jans-pycloudlib/jans/pycloudlib/lock/ldap_lock.py

@@ -12,6 +12,7 @@
 
 from jans.pycloudlib.lock.base_lock import BaseLock
 from jans.pycloudlib.utils import as_boolean
+from jans.pycloudlib.utils import get_password_from_file
 
 logger = logging.getLogger(__name__)
 
@@ -37,13 +38,11 @@ def __init__(self):
     def connection(self):
         user = "cn=Directory Manager"
 
-        password_file = os.environ.get("CN_LOCK_PASSWORD_FILE", "/etc/jans/conf/lock_password")
+        password_file = os.environ.get("CN_OCI_LOCK_PASSWORD_FILE", "/etc/jans/conf/oci_lock_password")
         if not os.path.isfile(password_file):
             password_file = os.environ.get("CN_LDAP_PASSWORD_FILE", "/etc/jans/conf/ldap_password")
 
-        with open(password_file) as f:
-            password = f.read().strip()
-
+        password = get_password_from_file(password_file)
         return Connection(self._server, user, password)
 
     def _prepare_schema(self):

jans-pycloudlib/jans/pycloudlib/lock/sql_lock.py

@@ -19,6 +19,7 @@
 from sqlalchemy.sql import select
 
 from jans.pycloudlib.lock.base_lock import BaseLock
+from jans.pycloudlib.utils import get_password_from_file
 
 if _t.TYPE_CHECKING:  # pragma: no cover
     # imported objects for function type hint, completion, etc.
@@ -50,12 +51,11 @@ def engine_url(self) -> str:
 
         user = os.environ.get("CN_SQL_DB_USER", "jans")
 
-        password_file = os.environ.get("CN_LOCK_PASSWORD_FILE", "/etc/jans/conf/lock_password")
+        password_file = os.environ.get("CN_OCI_LOCK_PASSWORD_FILE", "/etc/jans/conf/oci_lock_password")
         if not os.path.isfile(password_file):
             password_file = os.environ.get("CN_SQL_PASSWORD_FILE", "/etc/jans/conf/sql_password")
 
-        with open(password_file) as f:
-            password = f.read().strip()
+        password = get_password_from_file(password_file)
 
         if self._dialect in ("pgsql", "postgresql"):
             connector = "postgresql+psycopg2"

jans-pycloudlib/jans/pycloudlib/persistence/couchbase.py

@@ -26,6 +26,7 @@
 from jans.pycloudlib.utils import cert_to_truststore
 from jans.pycloudlib.utils import as_boolean
 from jans.pycloudlib.utils import safe_render
+from jans.pycloudlib.utils import get_password_from_file
 
 if _t.TYPE_CHECKING:  # pragma: no cover
     # imported objects for function type hint, completion, etc.
@@ -53,9 +54,8 @@ def _get_cb_password(manager: Manager, password_file: str, secret_name: str) ->
         Plaintext password.
     """
     if os.path.isfile(password_file):
-        with open(password_file) as f:
-            password = f.read().strip()
-            manager.secret.set(secret_name, password)
+        password = get_password_from_file(password_file)
+        manager.secret.set(secret_name, password)
     else:
         # get from secrets (if any)
         password = manager.secret.get(secret_name)

jans-pycloudlib/jans/pycloudlib/persistence/sql.py

@@ -24,6 +24,7 @@
 
 from jans.pycloudlib.utils import encode_text
 from jans.pycloudlib.utils import safe_render
+from jans.pycloudlib.utils import get_password_from_file
 
 if _t.TYPE_CHECKING:  # pragma: no cover
     # imported objects for function type hint, completion, etc.
@@ -52,9 +53,8 @@ def get_sql_password(manager: Manager) -> str:
     password_file = os.environ.get("CN_SQL_PASSWORD_FILE", "/etc/jans/conf/sql_password")
 
     if os.path.isfile(password_file):
-        with open(password_file) as f:
-            password = f.read().strip()
-            manager.secret.set(secret_name, password)
+        password = get_password_from_file(password_file)
+        manager.secret.set(secret_name, password)
     else:
         # get from secrets (if any)
         password = manager.secret.get(secret_name)

jans-pycloudlib/jans/pycloudlib/utils.py

@@ -1,8 +1,10 @@
 """This module contains various helpers."""
 
 import base64
+import binascii
 import json
 import logging
+import os
 import pathlib
 import random
 import re
@@ -22,6 +24,7 @@
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
 from ldap3.utils import hashed
+from sprig_aes import sprig_decrypt_aes
 
 from jans.pycloudlib.pki import generate_private_key
 from jans.pycloudlib.pki import generate_public_key
@@ -596,3 +599,58 @@ def generate_signed_ssl_certkey(
 
     sign_csr(cert_fn, csr, ca_key, ca_cert, valid_to=valid_to)
     return cert_fn, key_fn
+
+
+def get_password_from_file(password_file: str) -> str:
+    """Get password from file.
+
+    The contents of file will be loaded by the following priority:
+
+    1. Decode using AES CBC (sprig-aes implementation)
+    2. Decode using vanilla Base64
+    3. Plain text
+
+    Note that to decode using AES CBC, salt/key is loaded from file specified by
+    `CN_OCI_LOCK_SALT_FILE` environment variable (default to `/etc/jans/conf/oci_lock_salt`).
+
+    Args:
+        password_file: Path to file contains password.
+
+    Returns:
+        Plain text password.
+    """
+    with open(password_file) as f:
+        raw_passwd = f.read().strip()
+
+    salt_file = os.environ.get("CN_OCI_LOCK_SALT_FILE", "/etc/jans/conf/oci_lock_salt")
+
+    # probably sprig-aes format
+    if os.path.isfile(salt_file):
+        logger.info(f"Found salt file {salt_file} to decode password file {password_file}")
+        with open(salt_file) as f:
+            salt = f.read().strip()
+        try:
+            passwd = sprig_decrypt_aes(raw_passwd, salt).decode()
+            logger.info(f"Using sprig-aes to load password from {password_file}")
+        except ValueError as exc:
+            raise ValueError(
+                f"Unable to load password from {password_file} using sprig-aes "
+                f"(either {salt_file} or {password_file} is incompatible with sprig-aes); error={exc}"
+            )
+
+    # non sprig-aes format
+    else:
+        try:
+            # maybe vanilla base64
+            passwd = base64.b64decode(raw_passwd).decode()
+            logger.warning(f"Using base64 to load password from {password_file}")
+        except UnicodeDecodeError as exc:
+            # tried to decode from sprig-aes format
+            raise ValueError(f"Unable to load password from {password_file}; error={exc}")
+        except binascii.Error:
+            # fallback to plain text
+            passwd = raw_passwd
+            logger.warning(f"Using insecure method to load password from {password_file}")
+
+    # returns plain text
+    return passwd.strip()

jans-pycloudlib/setup.py

@@ -47,6 +47,7 @@ def find_version(*file_paths):
         # handle CVE-2022-36087
         "oauthlib>=3.2.1",
         "boto3",
+        "sprig-aes>=0.4.0",
     ],
     classifiers=[
         "Intended Audience :: Developers",

jans-pycloudlib/tests/test_utils.py

@@ -253,3 +253,52 @@ def test_generate_signed_ssl_certkey(tmpdir):
     assert os.path.isfile(str(base_dir.join("my-suffix.crt")))
     assert os.path.isfile(str(base_dir.join("my-suffix.csr")))
     assert os.path.isfile(str(base_dir.join("my-suffix.key")))
+
+
+@pytest.mark.parametrize("key, text, decrypted_text, sprig_enabled", [
+    # sprig-aes encoded
+    ("6Jsv61H7fbkeIkRvUpnZ98fu", "ow1Ty1OZWcOm8NRF49J07F1J1+fEQNLT5BKnCGqauvU=", "S3cr3t+pass", True),
+    # vanilla base64
+    ("6Jsv61H7fbkeIkRvUpnZ98fu", "UzNjcjN0K3Bhc3MK", "S3cr3t+pass", False),
+    # plain text
+    ("6Jsv61H7fbkeIkRvUpnZ98fu", "S3cr3t+pass", "S3cr3t+pass", False),
+])
+def test_get_password_from_file(monkeypatch, tmpdir, key, text, decrypted_text, sprig_enabled):
+    from jans.pycloudlib.utils import get_password_from_file
+
+    if sprig_enabled:
+        salt_file = tmpdir.join("oci_lock_salt")
+        salt_file.write(key)
+        monkeypatch.setenv("CN_OCI_LOCK_SALT_FILE", str(salt_file))
+
+    passwd_file = tmpdir.join("oci_lock_password")
+    passwd_file.write(text)
+
+    # ensure returning plain text password
+    assert get_password_from_file(str(passwd_file)) == decrypted_text
+
+
+def test_get_password_from_file_invalid_aes(monkeypatch, tmpdir):
+    from jans.pycloudlib.utils import get_password_from_file
+
+    salt_file = tmpdir.join("oci_lock_salt")
+    salt_file.write("6Jsv61H7fbkeIkRvUpnZ98fu")
+    monkeypatch.setenv("CN_OCI_LOCK_SALT_FILE", str(salt_file))
+
+    passwd_file = tmpdir.join("oci_lock_password")
+    passwd_file.write("S3cr3t+pass")
+
+    # ensure exception is thrown
+    with pytest.raises(ValueError):
+        get_password_from_file(str(passwd_file))
+
+
+def test_get_password_from_file_invalid_b64(monkeypatch, tmpdir):
+    from jans.pycloudlib.utils import get_password_from_file
+
+    passwd_file = tmpdir.join("oci_lock_password")
+    passwd_file.write("ow1Ty1OZWcOm8NRF49J07F1J1+fEQNLT5BKnCGqauvU=")
+
+    # ensure exception is thrown
+    with pytest.raises(ValueError):
+        get_password_from_file(str(passwd_file))