11from __future__ import annotations
22
3+ from base64 import b64decode
4+ import io
35import os
4- import logging
5- from typing import (IO , Optional ,Union )
6- from dotenv .main import load_dotenv as load_dotenv_file
6+ from typing import (IO , Optional , Union )
7+ from urllib .parse import urlparse , parse_qsl
78
8- from .vault import DotEnvVault
9-
10- logging .basicConfig (level = logging .INFO )
11-
12- logger = logging .getLogger (__name__ )
9+ from cryptography .hazmat .primitives .ciphers .aead import AESGCM
10+ from cryptography .exceptions import InvalidTag
11+ import dotenv .main as dotenv
1312
1413
1514def load_dotenv (
@@ -20,22 +19,137 @@ def load_dotenv(
2019 interpolate : bool = True ,
2120 encoding : Optional [str ] = "utf-8" ,
2221) -> bool :
22+ """This function will read your encrypted env file and load it
23+ into the environment for this process.
24+
25+ Call this function as close as possible to the start of your
26+ program (ideally in main).
27+
28+ If the `DOTENV_KEY` environment variable is set, `load_dotenv`
29+ will load encrypted environment settings from the `.env.vault`
30+ file in the current path.
31+
32+ If the `DOTENV_KEY` environment variable is not set, `load_dotenv`
33+ falls back to the behavior of the python-dotenv library, loading a
34+ specified (unencrypted) environment file.
35+
36+ Other parameters to `load_dotenv` are passed througg to the
37+ python-dotenv loader. In particular, whether `load_dotenv`
38+ overrides existing environment settings or not is determined by
39+ the `override` flag.
40+
2341 """
24- parameters are the same as python-dotenv library.
25- This is to inject the parameters to evironment variables.
26- """
27- dotenv_vault = DotEnvVault ()
28- if dotenv_vault . dotenv_key :
29- logger . info ( 'Loading env from encrypted .env.vault' )
30- vault_stream = dotenv_vault . parsed_vault ( dotenv_path = dotenv_path )
31- # we're going to override the .vault to any existing keys in local
32- return load_dotenv_file ( stream = vault_stream , override = True )
42+ if "DOTENV_KEY" in os . environ :
43+ vault_stream = parse_vault ( open ( ".env.vault" ))
44+ return dotenv . load_dotenv (
45+ dotenv_path = ".env.vault" ,
46+ stream = vault_stream ,
47+ verbose = verbose ,
48+ override = override ,
49+ interpolate = interpolate
50+ )
3351 else :
34- return load_dotenv_file (
52+ return dotenv . load_dotenv (
3553 dotenv_path = dotenv_path ,
3654 stream = stream ,
3755 verbose = verbose ,
3856 override = override ,
3957 interpolate = interpolate ,
4058 encoding = encoding
41- )
59+ )
60+
61+
62+ class DotEnvVaultError (Exception ):
63+ pass
64+
65+
66+ KEY_LENGTH = 64
67+
68+
69+ def parse_vault (vault_stream : io .IOBase ) -> io .StringIO :
70+ """Parse information from DOTENV_KEY, and decrypt vault.
71+ """
72+ dotenv_key = os .environ .get ("DOTENV_KEY" )
73+ if dotenv_key is None :
74+ raise DotEnvVaultError ("NOT_FOUND_DOTENV_KEY: Cannot find ENV['DOTENV_KEY']" )
75+
76+ # Use the python-dotenv library to read the .env.vault file.
77+ vault = dotenv .DotEnv (dotenv_path = ".env.vault" , stream = vault_stream )
78+
79+ # Extract segments from the DOTENV_KEY environment variable one by
80+ # one and retrieve the corresponding ciphertext from the vault
81+ # data.
82+ keys = []
83+ for dotenv_key_entry in [i .strip () for i in dotenv_key .split (',' )]:
84+ key , environment_key = parse_key (dotenv_key_entry )
85+
86+ ciphertext = vault .dict ().get (environment_key )
87+
88+ if not ciphertext :
89+ raise DotEnvVaultError (f"NOT_FOUND_DOTENV_ENVIRONMENT: Cannot locate environment { environment_key } )
90+
91+ keys .append ({
92+ 'encrypted_key' : key ,
93+ 'ciphertext' : ciphertext
94+ })
95+
96+ # Try decrypting environments one-by-one in the order they appear
97+ # in the DOTENV_KEY environment variable.
98+ decrypted = _key_rotation (keys = keys )
99+
100+ # Return the decrypted data as a text stream that we can pass to
101+ # the python-dotenv library.
102+ return io .StringIO (decrypted .decode ('utf-8' ))
103+
104+
105+ def parse_key (dotenv_key ):
106+ # Parse a single segment of the DOTENV_KEY environment variable.
107+ # These segments are in the form of URIs (see
108+ # https://www.dotenv.org/docs/security/dotenv-key).
109+ uri = urlparse (dotenv_key )
110+
111+ # The 64-character encryption key is stored in the password field
112+ # of the URI, possibly with a prefix.
113+ key = uri .password
114+ if len (key ) < KEY_LENGTH :
115+ raise DotEnvVault ('INVALID_DOTENV_KEY: Key part must be 64 characters long (or more)' )
116+
117+ # The environment is provided in the URI's query parameters.
118+ params = dict (parse_qsl (uri .query ))
119+ vault_environment = params .get ('environment' )
120+ if not vault_environment :
121+ raise DotEnvVaultError ('INVALID_DOTENV_KEY: Missing environment part' )
122+
123+ # Form the key used to store the ciphertext for this environment's
124+ # settings in the .env.vault file.
125+ environment_key = f'DOTENV_VAULT_{ vault_environment .upper ()}
126+
127+ return key , environment_key
128+
129+
130+ def _decrypt (ciphertext : str , key : str ) -> bytes :
131+ """decrypt method will decrypt via AES-GCM
132+ return: decrypted keys in bytes
133+ """
134+ # Remove any prefix from the encryption key (at this point, we
135+ # know that the key is at least 64 characters in length) and set
136+ # up the AES cipher.
137+ aesgcm = AESGCM (bytes .fromhex (key [- KEY_LENGTH :]))
138+
139+ # Decrypt the ciphertext: this is base64-encoded in the .env.vault
140+ # file, and the first 12 bytes of the decoded data are used as the
141+ # AES nonce value.
142+ ciphertext = b64decode (ciphertext )
143+ return aesgcm .decrypt (ciphertext [:12 ], ciphertext [12 :], b'' )
144+
145+
146+ def _key_rotation (keys : list [dict ]) -> str :
147+ """Iterate through list of keys to check for correct one.
148+ """
149+ for k in keys :
150+ try :
151+ return _decrypt (ciphertext = k ['ciphertext' ], key = k ['encrypted_key' ])
152+ except InvalidTag :
153+ continue
154+ raise DotEnvVaultError ('INVALID_DOTENV_KEY: Key must be valid.' )
155+
0 commit comments