Merge branch 'add-encryption'

roymanigley · roymanigley · commit 141d5fdf09da · 2025-03-25T20:22:58.000+01:00
diff --git a/requirements.txt b/requirements.txt
@@ -9,3 +9,4 @@ sh>=2
 tox
 twine
 wheel
+cryptography==44.0.2
diff --git a/setup.py b/setup.py
@@ -23,7 +23,7 @@ def read_files(files):
     version=meta["__version__"],
     author="Saurabh Kumar",
     author_email="me+github@saurabh-kumar.com",
-    url="https://github.com/theskumar/python-dotenv",
+    url="https://github.com/roymanigley/python-dotenv",
     keywords=[
         "environment variables",
         "deployments",
diff --git a/src/dotenv/cli.py b/src/dotenv/cli.py
@@ -5,6 +5,8 @@
 from contextlib import contextmanager
 from typing import Any, Dict, IO, Iterator, List, Optional
 
+from .crypt import Crypt
+
 try:
     import click
 except ImportError:
@@ -163,6 +165,24 @@ def run(ctx: click.Context, override: bool, commandline: List[str]) -> None:
     run_command(commandline, dotenv_as_dict)
 
 
+@cli.command(context_settings={'ignore_unknown_options': True})
+def get_secret_key() -> None:
+    try:
+        click.echo(Crypt.generate_key())
+    except Exception as e:
+        click.echo(f'Failed: {e}', err=True)
+
+
+@cli.command(context_settings={'ignore_unknown_options': True})
+@click.argument('value', type=str)
+@click.argument('secret_key', type=str)
+def encrypt(value: str, secret_key: str) -> None:
+    try:
+        click.echo(Crypt.encrypt_value(value, secret_key))
+    except Exception as e:
+        click.echo(f'Failed: {e}', err=True)
+
+
 def run_command(command: List[str], env: Dict[str, str]) -> None:
     """Replace the current process with the specified command.
 
diff --git a/src/dotenv/crypt.py b/src/dotenv/crypt.py
@@ -0,0 +1,47 @@
+import base64
+import re
+
+from cryptography.fernet import Fernet
+
+ENCRYPTION_REGEX = r'^enc{(.+)}$'
+
+
+class Crypt:
+
+    @classmethod
+    def is_encrypted_value(cls, value) -> bool:
+        return not not re.match(ENCRYPTION_REGEX, value)
+
+    @classmethod
+    def _extract_encrypted_value(cls, value) -> str:
+        return re.match(ENCRYPTION_REGEX, value).group(1)
+
+    @classmethod
+    def generate_key(cls) -> str:
+        key_bytes = Fernet.generate_key()
+        return base64.encodebytes(key_bytes).decode('utf-8')
+
+    @classmethod
+    def decrypt_value(cls, value: str, key: str) -> str:
+        if not key:
+            return value
+        key_bytes = base64.decodebytes(key.encode('utf-8'))
+        value_bytes = base64.decodebytes(cls._extract_encrypted_value(value).encode('utf-8'))
+        decrypted_bytes = Fernet(key=key_bytes).decrypt(value_bytes)
+        return decrypted_bytes.decode('utf-8')
+
+    @classmethod
+    def encrypt_value(cls, value: str, key: str) -> str:
+        key_bytes = base64.decodebytes(key.encode('utf-8'))
+        encrypted_bytes = Fernet(key=key_bytes).encrypt(value.encode('utf-8'))
+        return 'enc{' + base64.encodebytes(encrypted_bytes).decode('utf-8').strip().replace('\n', '') + '}'
+
+
+if __name__ == '__main__':
+    key = Crypt.generate_key()
+    message = 'Hello World'
+    encrypt_value = Crypt.encrypt_value(message, key)
+    print(encrypt_value)
+    print(Crypt.is_encrypted_value(encrypt_value))
+    print(Crypt._extract_encrypted_value(encrypt_value))
+    print(Crypt.decrypt_value(encrypt_value, key))
diff --git a/src/dotenv/main.py b/src/dotenv/main.py
@@ -40,6 +40,7 @@ def __init__(
         encoding: Optional[str] = None,
         interpolate: bool = True,
         override: bool = True,
+        secret_key: str = None
     ) -> None:
         self.dotenv_path: Optional[StrPath] = dotenv_path
         self.stream: Optional[IO[str]] = stream
@@ -48,6 +49,7 @@ def __init__(
         self.encoding: Optional[str] = encoding
         self.interpolate: bool = interpolate
         self.override: bool = override
+        self.secret_key = secret_key
 
     @contextmanager
     def _get_stream(self) -> Iterator[IO[str]]:
@@ -82,7 +84,7 @@ def dict(self) -> Dict[str, Optional[str]]:
 
     def parse(self) -> Iterator[Tuple[str, Optional[str]]]:
         with self._get_stream() as stream:
-            for mapping in with_warn_for_invalid_lines(parse_stream(stream)):
+            for mapping in with_warn_for_invalid_lines(parse_stream(stream, self.secret_key)):
                 if mapping.key is not None:
                     yield mapping.key, mapping.value
 
@@ -156,6 +158,7 @@ def set_key(
     quote_mode: str = "always",
     export: bool = False,
     encoding: Optional[str] = "utf-8",
+    secret_key: str = None,
 ) -> Tuple[Optional[bool], str, str]:
     """
     Adds or Updates a key/value to the given .env
@@ -182,7 +185,7 @@ def set_key(
     with rewrite(dotenv_path, encoding=encoding) as (source, dest):
         replaced = False
         missing_newline = False
-        for mapping in with_warn_for_invalid_lines(parse_stream(source)):
+        for mapping in with_warn_for_invalid_lines(parse_stream(source, secret_key)):
             if mapping.key == key_to_set:
                 dest.write(line_out)
                 replaced = True
@@ -202,6 +205,7 @@ def unset_key(
     key_to_unset: str,
     quote_mode: str = "always",
     encoding: Optional[str] = "utf-8",
+    secret_key: str = None
 ) -> Tuple[Optional[bool], str]:
     """
     Removes a given key from the given `.env` file.
@@ -215,7 +219,7 @@ def unset_key(
 
     removed = False
     with rewrite(dotenv_path, encoding=encoding) as (source, dest):
-        for mapping in with_warn_for_invalid_lines(parse_stream(source)):
+        for mapping in with_warn_for_invalid_lines(parse_stream(source, secret_key)):
             if mapping.key == key_to_unset:
                 removed = True
             else:
@@ -329,6 +333,7 @@ def load_dotenv(
     override: bool = False,
     interpolate: bool = True,
     encoding: Optional[str] = "utf-8",
+    secret_key: Optional[str] = os.environ.get('DOTENV_SECRET'),
 ) -> bool:
     """Parse a .env file and then load all the variables found as environment variables.
 
@@ -358,6 +363,7 @@ def load_dotenv(
         interpolate=interpolate,
         override=override,
         encoding=encoding,
+        secret_key=secret_key
     )
     return dotenv.set_as_environment_variables()
 
@@ -368,6 +374,7 @@ def dotenv_values(
     verbose: bool = False,
     interpolate: bool = True,
     encoding: Optional[str] = "utf-8",
+    secret_key: Optional[str] = None
 ) -> Dict[str, Optional[str]]:
     """
     Parse a .env file and return its content as a dict.
@@ -395,4 +402,5 @@ def dotenv_values(
         interpolate=interpolate,
         override=True,
         encoding=encoding,
+        secret_key=secret_key
     ).dict()
diff --git a/src/dotenv/parser.py b/src/dotenv/parser.py
@@ -3,6 +3,8 @@
 from typing import (IO, Iterator, Match, NamedTuple, Optional,  # noqa:F401
                     Pattern, Sequence, Tuple)
 
+from dotenv.crypt import Crypt
+
 
 def make_regex(string: str, extra_flags: int = 0) -> Pattern[str]:
     return re.compile(string, re.UNICODE | extra_flags)
@@ -118,21 +120,25 @@ def parse_unquoted_value(reader: Reader) -> str:
     return re.sub(r"\s+#.*", "", part).rstrip()
 
 
-def parse_value(reader: Reader) -> str:
+def parse_value(reader: Reader, secret_key: str) -> str:
     char = reader.peek(1)
     if char == u"'":
         (value,) = reader.read_regex(_single_quoted_value)
-        return decode_escapes(_single_quote_escapes, value)
+        value = decode_escapes(_single_quote_escapes, value)
     elif char == u'"':
         (value,) = reader.read_regex(_double_quoted_value)
-        return decode_escapes(_double_quote_escapes, value)
+        value = decode_escapes(_double_quote_escapes, value)
     elif char in (u"", u"\n", u"\r"):
-        return u""
+        value = u""
     else:
-        return parse_unquoted_value(reader)
+        value = parse_unquoted_value(reader)
+
+    if Crypt.is_encrypted_value(value):
+        return Crypt.decrypt_value(value, secret_key)
+    return value
 
 
-def parse_binding(reader: Reader) -> Binding:
+def parse_binding(reader: Reader, secret_key: str = None) -> Binding:
     reader.set_mark()
     try:
         reader.read_regex(_multiline_whitespace)
@@ -148,7 +154,7 @@ def parse_binding(reader: Reader) -> Binding:
         reader.read_regex(_whitespace)
         if reader.peek(1) == "=":
             reader.read_regex(_equal_sign)
-            value: Optional[str] = parse_value(reader)
+            value: Optional[str] = parse_value(reader, secret_key)
         else:
             value = None
         reader.read_regex(_comment)
@@ -168,8 +174,13 @@ def parse_binding(reader: Reader) -> Binding:
             error=True,
         )
 
+def parse_secret_value(value: str) -> str:
+
+    return value
+
+
 
-def parse_stream(stream: IO[str]) -> Iterator[Binding]:
+def parse_stream(stream: IO[str], secret_key: str = None) -> Iterator[Binding]:
     reader = Reader(stream)
     while reader.has_next():
-        yield parse_binding(reader)
+        yield parse_binding(reader, secret_key)
diff --git a/tests/test_secret_key.py b/tests/test_secret_key.py
@@ -0,0 +1,30 @@
+import dotenv
+from dotenv.crypt import Crypt
+
+
+def test_dotenv_with_secret_values__should_not_encrypt__due_to_no_key(dotenv_path):
+    secret_key = Crypt.generate_key()
+    secret_value = 'my super secret secret'
+    secret_value_decrypted = Crypt.encrypt_value(secret_value, secret_key)
+    assert secret_value_decrypted != secret_value
+
+    dotenv_path.write_text(f'my_secret={secret_value_decrypted}')
+
+    with dotenv_path.open() as f:
+        result = dotenv.dotenv_values(stream=f)
+
+    assert result == {"my_secret": secret_value_decrypted}
+
+
+def test_dotenv_with_secret_values__should_encrypt(dotenv_path):
+    secret_key = Crypt.generate_key()
+    secret_value = 'my super secret secret'
+    secret_value_decrypted = Crypt.encrypt_value(secret_value, secret_key)
+    assert secret_value_decrypted != secret_value
+
+    dotenv_path.write_text(f'my_secret={secret_value_decrypted}')
+
+    with dotenv_path.open() as f:
+        result = dotenv.dotenv_values(stream=f, secret_key=secret_key)
+
+    assert result == {"my_secret": secret_value}

-Original file line number
+Diff line change
 tox
 twine
 wheel
 +cryptography==44.0.2