EVX-3 Add encryption and stream support

Introduced AES-256 encryption support for `.env` files, enabling better support for secure handling of sensitive data.
Added `envcrypt` CLI utility to encrypt and decrypt `.env` files, with corresponding tests for validation.
Updated the `Env` class and associated modules to support decryption, improved environment variable loading, and adjusted Python requirements to >=3.11.

.envrc

@@ -1,3 +1,5 @@
-layout virtualenv .venv
+layout uv
+source_up
 dotenv_if_exists
 PATH_add scripts
+export TEST_PASSWORD="jV4cl:aPx2D9s"

.gitignore

@@ -301,4 +301,4 @@ Session.vim
 .env
 
 requirements.txt
-output.env
+test.env*

CHANGELOG.md

@@ -1,5 +1,20 @@
 # ChangeLog
 
+### v4.0.0
+
+- :warning: BREAKING CHANGE: kwargs passed to Env() are no longer added to the env if readenv=False. This is probably of no consequence as it is a (mis?)feature that rarely (if ever) was used.
+- Bugfix: dicts passed in *args the Env() are correctly converted to str->str mappings.
+- Feature: Env can now take BytesIO and StringIO objects in Env(*args). Since these are immediate objects, they are handled as priority variables, different to variables set via `.env` files in that they overwrite existing variables by default. Explicitly using the overwrite=False changes this behaviour.
+- Warning: To provide support for different types of streams, environment files are now handled internally as bytes, however before evaluation they are converted via an encoding parameter that defaults to utf-8.
+- Feature: Encrypted `.env` files (`.env.enc`) are now supported, meaing that you don't need to implement a Hashicorp Vault in order to avoid having plain text secrets on the filesystem, you can simply encrypt the `.env` file directly]. Use the `decrypt=True` parameter and provide - directly or indirectly - the encryption password used to derive the key:
+
+  - `password=<password_value>`
+  - `password=$<environment_variable_name>`
+  - `password=/<filename to read>`
+- If `decrypt=True` is used with a password (see previous item) `envex` will look for `.env.enc` (or more correctly `${DOTENV:-.env}.enc`) and use that if it exists. Regardless, encryped content is supported in both `.env` and `.env.enc` files, but the `.enc` file is used in preference. Encrypted files have a special initial 4-byte signature that distinguishes them from unencrypted files.
+- A utility cli script `envcrypt` is provided to support both encryption and decryption.
+  Use `envcrypt -h` for usage.
+
 ### v3.2.0
 
 - upgrade dependencies (several) including vulnerability fixes

README.md

@@ -7,23 +7,21 @@
 
 ## Overview
 
-This module provides a convenient interface for handling the environment, and therefore configuration of any application
-using 12factor.net principals removing many environment-specific variables and security sensitive information from
-application code.
-
-An `Env` instance delivers a lot of functionality by providing a type-smart front-end to `os.environ`,
-providing a superset of `os.environ` functionality, including setting default values.
-
-From version 2.0, this module also supports transparently fetching values from Hashicorp vault,
-which reduces the need to store secrets in plain text on the filesystem.
-This functionality is optional, activated automatically when the `hvac` module is installed, and connection and
-authentication to Vault succeed.
-Only get (no set) operations to Vault are supported.
+This module provides a convenient interface for handling the environment, and therefore configuration of any application using 12factor.net principals removing many environment-specific variables and security-sensitive information from application code.
+
+An `Env` instance delivers a lot of functionality by providing a type-smart front-end to `os.environ`, providing a superset of `os.environ` functionality, including setting default values.
+
+`envex` supports AES-256 encrypted environment files, and if enabled by using the `decrypt=True` argument and providing the decryption password searches for `.env.enc` files first but falls back to `.env`.
+This avoids having plain text files on the filesystem that contain sensitive information.
+The provided `env_crypto` utility allows conversion between encrypted and non-encrypted formats.
+
+Alternatively, `envex` handles transparently fetching values from Hashicorp vault, reducing the need to store secrets in plain text on the filesystem and depending on the environment may present a more convenient way of managing application secrets.
+Hashicorp vault functionality is optional, activated automatically when the `hvac` module is installed into the active virtual environment, and connection and authentication to Vault succeed.
 Values fetched from Vault are cached by default to reduce the overhead of the api call.
-If this is of concern to security, caching can be disabled using the `enable_cache=False` parameter to Env.
+If this is of concern to security, caching can be disabled using the`enable_cache=False` parameter to Env.
 
-This module provides some features not supported by other dotenv handlers (python-dotenv, etc.) including recursive
-expansion of template variables, which can be very useful for DRY.
+This module provides man y features not supported by other dotenv handlers (python-dotenv, etc.) including recursive
+expansion of template variables, supporting the don't-repeat-yourself (DRY) principle.
 
 ```python
 from envex import env
@@ -148,11 +146,12 @@ The SecretsManager and Vault client leverage environment variables for their con
 This ensures a degree of transparency as it allows the client to use them but mitigates the need for the client code to be aware of their presence.
 A summary of these variables is in the following table:
 
-| Variable          | Description                                                              |
-|-------------------|--------------------------------------------------------------------------|
-| VAULT_ADDR        | The URL of the vault server                                              |
-| VAULT_TOKEN       | The vault token to use for authentication                                |
-| VAULT_CACERT      | The path to the CA certificate to use for TLS verification               |
-| VAULT_CAPATH      | The path to a directory of CA certificates to use for TLS verification   |
-| VAULT_CLIENT_CERT | The path to the client certificate to use for TLS connection             |
-| VAULT_CLIENT_KEY  | The path to the client key to use for TLS connection                     |
+
+| Variable          | Description                                                            |
+| ----------------- | ---------------------------------------------------------------------- |
+| VAULT_ADDR        | The URL of the vault server                                            |
+| VAULT_TOKEN       | The vault token to use for authentication                              |
+| VAULT_CACERT      | The path to the CA certificate to use for TLS verification             |
+| VAULT_CAPATH      | The path to a directory of CA certificates to use for TLS verification |
+| VAULT_CLIENT_CERT | The path to the client certificate to use for TLS connection           |
+| VAULT_CLIENT_KEY  | The path to the client key to use for TLS connection                   |

envex/dot_env.py

@@ -2,18 +2,26 @@
 import os
 import sys
 import contextlib
+from io import TextIOBase, BytesIO
 from pathlib import Path
 from string import Template
-from typing import List, MutableMapping, Union
+from typing import Dict, List, MutableMapping, Union, Optional, ContextManager, BinaryIO
+
+from .env_crypto import decrypt_data, DecryptError
 
 __all__ = (
     "load_env",
+    "load_stream",
     "load_dotenv",  # alias
+    "update_env",
     "unquote",
 )
 
+
 DEFAULT_ENVKEY = "DOTENV"
 DEFAULT_DOTENV = ".env"
+ENCRYPTED_EXT = ".enc"
+DEFAULT_ENCODING = "utf-8"
 
 
 def unquote(line, quotes="\"'"):
@@ -22,6 +30,11 @@ def unquote(line, quotes="\"'"):
     return line
 
 
+def update_env(env: MutableMapping[str, str], mapping: Dict):
+    for k, v in mapping.items():
+        env[str(k)] = str(v)
+
+
 def _env_default(
     environ: MutableMapping[str, str], key: str, val: str, overwrite: bool = False
 ):
@@ -38,36 +51,48 @@ def _env_export(
 
 
 def _env_files(
-    env_file: str, search_path: List[Path], parents: bool, errors: bool
+    env_file: str, search_path: List[Path], parents: bool, decrypt: bool, errors: bool
 ) -> List[str]:
     """expand env_file with the full search path, optionally parents as well"""
-
     searched = []
+
+    def search_dotpath(base: Path, name: str):
+        _path = os.path.join(base, name)
+        return _path if os.access(_path, os.R_OK) else None
+
     for path in search_path:
         path = path.resolve()
         if not path.is_dir():
             path = path.parent
         searched.append(path)
         paths = [path] + list(path.parents)
+
         # search a path and parents
         for sub_path in paths:
-            # stop at first found, or ...
-            # fail fast unless searching parents
-            env_path = os.path.join(sub_path, env_file)
-            if os.access(env_path, os.R_OK):
+            # if decryption is enabled, encrypted .env.enc files take priority
+            env_path = (
+                search_dotpath(sub_path, env_file + ENCRYPTED_EXT) if decrypt else None
+            )
+            if env_path:
                 yield env_path
-            elif not parents:
-                break
+            else:
+                # but allow fallback to standard .env
+                env_path = search_dotpath(sub_path, env_file)
+                if env_path:
+                    yield env_path
+                # quit search unless we are searching up
+                elif not parents:
+                    break
     if errors:
         raise FileNotFoundError(f"{env_file} in {[s.as_posix() for s in searched]}")
     else:
         yield env_file
 
 
 @contextlib.contextmanager
-def open_env(path: Union[str, Path]):
+def open_env(path: Union[str, Path]) -> ContextManager[BinaryIO]:
     """same as open, allow monkeypatch"""
-    fp = open(path, "r")
+    fp = open(path, "rb")
     try:
         yield fp
     finally:
@@ -79,6 +104,41 @@ def open_env(path: Union[str, Path]):
 }
 
 
+def _process_line(_lineno: int, string: str, errors: bool, _env_path: Path | None):
+    """process a single line"""
+    _func, _key, _val = _env_default, None, None
+    parts = string.split("=", 1)
+    if len(parts) == 2:
+        _key, _val = parts
+    elif len(parts) == 1:
+        _key = parts[0]
+    if _key:
+        words = _key.split(maxsplit=1)
+        if len(words) > 1:
+            command, _key = words
+            try:
+                _func = ENV_COMMANDS[command]
+            except KeyError:
+                if errors:
+                    path = _env_path.as_posix() if _env_path else "stream"
+                    print(
+                        f"unknown command {command} {path}({_lineno})",
+                        file=sys.stderr,
+                    )
+    return _func, unquote(_key), unquote(_val)
+
+
+def _process_stream(
+    stream: BytesIO, environ, overwrite, errors, encoding=DEFAULT_ENCODING, env_path=None
+):
+    for lineno, line in enumerate(stream.readlines(), start=1):
+        line = line.decode(encoding).strip()
+        if line and line[0] != "#":
+            func, key, val = _process_line(lineno, line, errors, env_path)
+            if func is not None:
+                func(environ, key, val, overwrite=overwrite)
+
+
 def _process_env(
     env_file: str,
     search_path: List[Path],
@@ -87,6 +147,9 @@ def _process_env(
     parents: bool,
     errors: bool,
     working_dirs: bool,
+    decrypt: bool,
+    password: Optional[str] = None,
+    encoding: str = DEFAULT_ENCODING,
 ) -> MutableMapping[str, str]:
     """
     search for any env_files in the given dir list and populate environ dict
@@ -96,50 +159,32 @@ def _process_env(
     :param environ: environment to update
     :param overwrite: whether to overwrite existing values
     :param parents: whether to search upwards until a file is found
+    :param decrypt: whether to attempt decryption
     :param errors: whether to raise FileNotFoundError if the env_file is not found
     :param working_dirs: whether to add the env file's directory
+    :param encoding: text encoding
     """
 
-    def process_line(_env_path: Path, _lineno: int, string: str):
-        """process a single line"""
-        _func, _key, _val = _env_default, None, None
-        parts = string.split("=", 1)
-        if len(parts) == 2:
-            _key, _val = parts
-        elif len(parts) == 1:
-            _key = parts[0]
-        if _key:
-            words = _key.split(maxsplit=1)
-            if len(words) > 1:
-                command, _key = words
-                try:
-                    _func = ENV_COMMANDS[command]
-                except KeyError:
-                    if errors:
-                        print(
-                            f"unknown command {command} {_env_path.as_posix()}({_lineno})",
-                            file=sys.stderr,
-                        )
-        return _func, unquote(_key), unquote(_val)
-
     files_not_found = []
     files_found = False
-    for env_path in _env_files(env_file, search_path, parents, errors):
+    for env_path in _env_files(env_file, search_path, parents, decrypt, errors):
         # insert PWD as container of env file
         env_path = Path(env_path).resolve()
         if working_dirs:
             environ["PWD"] = str(env_path.parent)
         try:
             with open_env(env_path) as f:
-                lineno = 0
-                for line in f.readlines():
-                    line = line.strip()
-                    lineno += 1
-                    if line and line[0] != "#":
-                        func, key, val = process_line(env_path, lineno, line)
-                        if func is not None:
-                            func(environ, key, val, overwrite=overwrite)
-                files_found = True
+                load_stream(
+                    BytesIO(f.read()),
+                    environ,
+                    overwrite,
+                    errors,
+                    decrypt,
+                    password,
+                    encoding,
+                    env_path,
+                )
+            files_found = True
         except FileNotFoundError:
             files_not_found.append(env_path)
     if errors and not files_found and files_not_found:
@@ -180,17 +225,23 @@ def load_env(
     update: bool = True,
     errors: bool = False,
     working_dirs: bool = True,
+    decrypt: bool = False,
+    password: str = None,
+    encoding: Optional[str] = DEFAULT_ENCODING,
 ) -> MutableMapping[str, str]:
     """
     Loads one or more .env files with optional nesting, updating os.environ
     :param env_file: name of the environment file (.env or $ENV default)
     :param search_path: single or list of directories in order of precedence - str, bytes or Path
+    :param environ: environment mapping to process
     :param overwrite: whether to overwrite existing values
     :param parents: whether to search upwards until a file is found
     :param update: option to update os.environ, default=True
     :param errors: whether to raise FileNotFoundError if env_file not found
     :param working_dirs: whether to add the env file's directory
-    :param environ: environment mapping to process
+    :param decrypt: whether to support encrypted .env.enc
+    :param password: decryption password
+    :param encoding: text encoding (default utf-8)
     :returns the new environment
     """
     if environ is None:
@@ -229,10 +280,32 @@ def load_env(
             parents,
             errors,
             working_dirs,
+            decrypt,
+            password,
+            encoding,
         )
     )
     # optionally update the actual environment
     return _update_os_env(environ) if update else environ
 
 
+def load_stream(
+    stream: Union[BytesIO, TextIOBase],
+    environ: MutableMapping[str, str] = None,
+    overwrite: bool = False,
+    errors: bool = False,
+    decrypt: bool = False,
+    password: Optional[str] = None,
+    encoding: Optional[str] = DEFAULT_ENCODING,
+    env_path: Optional[Path] = None,
+):
+    if isinstance(stream, TextIOBase):
+        stream.seek(0)
+        stream = BytesIO(stream.read().encode(encoding))
+    elif password and decrypt:
+        with contextlib.suppress(DecryptError):
+            stream = decrypt_data(stream, password)
+    _process_stream(stream, environ, overwrite, errors, encoding, env_path)
+
+
 load_dotenv = load_env