1"""Bash-compatible `.env` file parsing and Plain dev/test dotenv loading.
2
3`plain.dev` owns all dotenv code so that production deployments (which don't
4install plain.dev) never load `.env` files. plain.pytest opportunistically
5imports `load_dotenv_files` — if plain.dev is installed, `.env.test*` loads
6under pytest; if not, the plugin skips dotenv loading entirely.
7
8Parser supports:
9- KEY=value (basic unquoted)
10- KEY="double quoted value" (with escape handling and multiline)
11- KEY='single quoted value' (literal, including multiline)
12- export KEY=value (strips export prefix)
13- Comments (# comment and inline KEY=value # comment)
14- Variable expansion: $VAR and ${VAR} (in unquoted and double-quoted values)
15- Command substitution: $(command)
16- Encrypted values: KEY=encrypted:<token>, decrypted with DEV_ENV_KEY
17
18`encrypted:` is recognized syntactically — at the start of an *unquoted* value,
19before anything is expanded. So an encrypted value is never expanded, and
20quoting it (`KEY='encrypted:aes'`) makes it ordinary text.
21
22Encrypted values are resolved in a second phase, after parsing. `load_dotenv_files`
23resolves once after every file has loaded, so the key line can live in
24`.env.dev.local`, in the shell, or as a committed `DEV_ENV_KEY=$(op read ...)`
25reference next to the values themselves. Decrypted plaintext is bound literally —
26no variable expansion or command substitution.
27"""
28
29from __future__ import annotations
30
31import os
32import re
33import subprocess
34from collections.abc import Callable
35from pathlib import Path
36from typing import NamedTuple
37
38import click
39from plain.exceptions import ImproperlyConfigured
40
41__all__ = ["load_dotenv", "load_dotenv_files", "parse_dotenv"]
42
43# Environment variable holding the project's Fernet key.
44ENV_KEY_VAR = "DEV_ENV_KEY"
45# Written form of an encrypted value: `KEY=encrypted:<fernet token>`.
46ENCRYPTED_VALUE_PREFIX = "encrypted:"
47
48# Match ${VAR} or $VAR (VAR must start with letter/underscore, then alphanumeric/underscore)
49_VAR_BRACE_RE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
50_VAR_BARE_RE = re.compile(r"\$([A-Za-z_][A-Za-z0-9_]*)")
51# Placeholder for escaped $ (to prevent expansion)
52_ESCAPED_DOLLAR = "\x00DOLLAR\x00"
53
54_PLAIN_ENV_RE = re.compile(r"^[a-zA-Z][a-zA-Z0-9_-]*$")
55_files_loaded = False
56
57# Which file bound each key, filled in as files load. A key already in the
58# environment before any file loaded (exported in the shell) is simply absent.
59# `plain env set` reads this to warn about a name it can't actually change.
60bound_sources: dict[str, Path] = {}
61
62
63class Binding(NamedTuple):
64 """One `KEY=value` binding, and where it sits in the content it came from.
65
66 `key_start` is the first character of the key, so any `export ` prefix and
67 indentation are kept, and `value_end` is just past the value, so an inline
68 comment, the line ending, or a further binding on the same line is kept too.
69 `value` is the value as the parser read it — quotes removed and escapes
70 processed, and expanded only if the parser was expanding.
71 """
72
73 key: str
74 value: str
75 encrypted: bool
76 source: Path | None
77 key_start: int
78 value_end: int
79
80
81# --- the ladder and the loaders ---
82
83
84def dotenv_ladder(plain_env: str) -> list[str]:
85 """The `.env` files that load, highest precedence first.
86
87 1. `.env.{plain_env}.local` — gitignored, env-specific personal
88 2. `.env.local` — gitignored, personal; SKIPPED in test
89 3. `.env.{plain_env}` — gitignored or committed, env-specific
90 4. `.env` — committed baseline
91 """
92 paths = []
93 if plain_env:
94 paths.append(f".env.{plain_env}.local")
95 if plain_env != "test":
96 # Skipped under test (Next.js / Rails dotenv convention) so CI runs
97 # stay deterministic and personal creds don't leak into the suite.
98 paths.append(".env.local")
99 if plain_env:
100 paths.append(f".env.{plain_env}")
101 paths.append(".env")
102 return paths
103
104
105def load_dotenv_files(*, decrypt: bool = True) -> None:
106 """Load `.env` files using Next.js / Vite-style precedence.
107
108 Files load in `dotenv_ladder()` order — `load_dotenv()` doesn't override
109 existing keys, so the first file to bind a key wins.
110
111 `PLAIN_ENV` is set by the CLI dispatcher (`plain.cli.core`) based on the
112 active command — `plain dev` → `dev`, `plain test` → `test` — and by
113 `plain env`, which sets its own. Export `PLAIN_ENV` yourself to override.
114
115 With `decrypt=False`, plain values bind as usual and encrypted ones are
116 left unbound: no key is needed, and nothing fails. That's what `plain env`
117 loads with, since it's the command you run when decryption can't work yet.
118
119 Idempotent within a process — repeat calls are a no-op.
120 """
121 global _files_loaded
122 if _files_loaded:
123 return
124 _files_loaded = True
125
126 plain_env = os.environ.get("PLAIN_ENV", "")
127 if plain_env and not _PLAIN_ENV_RE.fullmatch(plain_env):
128 raise ValueError(
129 f"PLAIN_ENV must match {_PLAIN_ENV_RE.pattern}, got {plain_env!r}"
130 )
131
132 bound_sources.clear()
133
134 # Encrypted values from every file are collected here and decrypted once
135 # at the end, so DEV_ENV_KEY can come from any file (or the shell).
136 deferred: dict[str, Binding] = {}
137
138 for path in dotenv_ladder(plain_env):
139 if _load_dotenv_deferring_encrypted(path, override=False, deferred=deferred):
140 click.secho(f"Loading {path}...", dim=True, italic=True, err=True)
141
142 if decrypt:
143 _bind_decrypted(deferred)
144
145
146def load_dotenv(
147 filepath: str | Path,
148 *,
149 override: bool = False,
150 decrypt: bool = True,
151) -> bool:
152 """
153 Load environment variables from a .env file into os.environ.
154
155 Args:
156 filepath: Path to the .env file
157 override: If True, overwrite existing environment variables
158 decrypt: If False, leave encrypted values unbound instead of decrypting
159
160 Returns:
161 True if the file was loaded, False if it doesn't exist
162
163 Encrypted values are decrypted with DEV_ENV_KEY once the whole file has
164 been parsed, so the key may be defined earlier in the same file.
165 """
166 deferred: dict[str, Binding] = {}
167 loaded = _load_dotenv_deferring_encrypted(
168 filepath, override=override, deferred=deferred
169 )
170 if not loaded:
171 return False
172 if decrypt:
173 _bind_decrypted(deferred)
174 return True
175
176
177def parse_dotenv(filepath: str | Path, *, decrypt: bool = True) -> dict[str, str]:
178 """
179 Parse a .env file and return a dictionary of key-value pairs.
180
181 Does not modify os.environ. Supports multiline values in quoted strings.
182 Encrypted values are decrypted with DEV_ENV_KEY from os.environ, falling
183 back to a `DEV_ENV_KEY=` line in the file itself.
184 """
185 path = Path(filepath)
186 content = path.read_text(encoding="utf-8")
187
188 encrypted: list[Binding] = []
189
190 def collect(binding: Binding) -> None:
191 if binding.encrypted:
192 encrypted.append(binding)
193
194 result = _FileParser(content, source=path).parse(collect)
195 if not decrypt:
196 return result
197
198 env_key = os.environ.get(ENV_KEY_VAR) or result.get(ENV_KEY_VAR)
199 for binding in encrypted:
200 result[binding.key] = decrypt_env_binding(binding, env_key=env_key)
201 return result
202
203
204def _load_dotenv_deferring_encrypted(
205 filepath: str | Path,
206 *,
207 override: bool,
208 deferred: dict[str, Binding],
209) -> bool:
210 """Bind a file's plain values now and collect its encrypted ones in `deferred`.
211
212 A key with a deferred value counts as bound: a later file (or a later line)
213 can't take it over, which keeps first-file-wins precedence intact.
214 """
215 path = Path(filepath)
216 if not path.exists():
217 return False
218
219 content = path.read_text(encoding="utf-8")
220
221 # Keys that are already bound: their values are located but not expanded,
222 # so nothing in them runs and nothing in them is decrypted.
223 skip_for = None if override else set(os.environ) | deferred.keys()
224
225 def on_bind(binding: Binding) -> None:
226 key = binding.key
227 if not override and (key in os.environ or key in deferred):
228 return
229 if override:
230 deferred.pop(key, None)
231 if binding.encrypted:
232 deferred[key] = binding
233 else:
234 os.environ[key] = binding.value
235 bound_sources[key] = path
236
237 _FileParser(
238 content,
239 source=path,
240 skip_for=skip_for,
241 encrypted_names=set(deferred),
242 ).parse(on_bind)
243 return True
244
245
246def _bind_decrypted(deferred: dict[str, Binding]) -> None:
247 """Decrypt every deferred binding with DEV_ENV_KEY and bind the plaintext literally."""
248 for binding in deferred.values():
249 os.environ[binding.key] = decrypt_env_binding(binding)
250
251
252# --- encrypted values ---
253
254
255def decrypt_env_binding(binding: Binding, env_key: str | None = None) -> str:
256 """Decrypt one binding, or raise ImproperlyConfigured saying exactly what is wrong."""
257 from cryptography.fernet import InvalidToken
258
259 if env_key is None:
260 env_key = os.environ.get(ENV_KEY_VAR, "")
261
262 if not env_key:
263 if ENV_KEY_VAR in os.environ:
264 raise ImproperlyConfigured(
265 f"{binding.key} in {binding.source} is encrypted, but {ENV_KEY_VAR} is "
266 f"set to an empty value. If it is bound with a command like "
267 f'{ENV_KEY_VAR}=$(op read "op://..."), that command failed — run it '
268 "yourself to see why."
269 )
270 plain_env = os.environ.get("PLAIN_ENV", "")
271 key_file_hint = f".env.{plain_env}.local" if plain_env else ".env.local"
272 raise ImproperlyConfigured(
273 f"{binding.key} in {binding.source} is encrypted, but {ENV_KEY_VAR} is not set. "
274 f"Set {ENV_KEY_VAR} to this project's key — usually with a line in {key_file_hint} "
275 f'like {ENV_KEY_VAR}=$(op read "op://..."), or commit that reference line in '
276 f"{binding.source} so teammates get it too. Generate a new key with `plain env key`."
277 )
278
279 try:
280 return decrypt_env_value(binding.value, env_key)
281 except (InvalidToken, ValueError) as e:
282 raise ImproperlyConfigured(
283 f"{ENV_KEY_VAR} does not decrypt {binding.key} in {binding.source}. "
284 "Check that it is this project's key. If the value is meant to be the "
285 f"literal text and not an encrypted value, quote it: "
286 f"{binding.key}='{ENCRYPTED_VALUE_PREFIX}...'"
287 ) from e
288
289
290def is_encrypted_value(value: str) -> bool:
291 """True when a value is written in the `encrypted:<token>` form.
292
293 A syntactic test on the text as written — an unquoted value starting with
294 `encrypted:` is an encrypted value, and a token that doesn't decrypt is an
295 error rather than plain text.
296 """
297 return value.startswith(ENCRYPTED_VALUE_PREFIX)
298
299
300def generate_env_key() -> str:
301 """Generate a new DEV_ENV_KEY (a Fernet key: urlsafe base64, 44 chars)."""
302 from cryptography.fernet import Fernet
303
304 return Fernet.generate_key().decode("ascii")
305
306
307def encrypt_env_value(plaintext: str, env_key: str) -> str:
308 """Encrypt a plaintext string into the `encrypted:<token>` form."""
309 from cryptography.fernet import Fernet
310
311 token = Fernet(env_key.encode("ascii")).encrypt(plaintext.encode("utf-8"))
312 return ENCRYPTED_VALUE_PREFIX + token.decode("ascii")
313
314
315def decrypt_env_value(value: str, env_key: str) -> str:
316 """Decrypt an `encrypted:<token>` value back to its plaintext string.
317
318 Raises `cryptography.fernet.InvalidToken` when the key doesn't match or the
319 token is malformed, and `ValueError` when `value` isn't in the encrypted
320 form or the key itself is malformed.
321 """
322 from cryptography.fernet import Fernet
323
324 if not is_encrypted_value(value):
325 raise ValueError(f"Not an {ENCRYPTED_VALUE_PREFIX} value")
326 token = value.removeprefix(ENCRYPTED_VALUE_PREFIX).encode("ascii")
327 return Fernet(env_key.encode("ascii")).decrypt(token).decode("utf-8")
328
329
330# --- finding a binding to rewrite ---
331
332
333def find_env_binding(
334 content: str, name: str, *, source: Path | None = None
335) -> Binding | None:
336 """Find the first `NAME=...` binding in .env file content, without evaluating anything.
337
338 This runs the loader's own parser in raw mode, so what counts as a binding
339 here — `export NAME=`, whitespace around `=`, quoted values spanning lines,
340 a second binding on the same line — is exactly what counts when the file is
341 loaded. `source` is only recorded on the binding, for error messages.
342 """
343 found: Binding | None = None
344
345 def match(binding: Binding) -> None:
346 nonlocal found
347 if found is None and binding.key == name:
348 found = binding
349
350 _FileParser(content, source=source, expand=False).parse(match)
351 return found
352
353
354# --- the parser ---
355
356
357class _FileParser:
358 """Parses the content of one .env file, top to bottom.
359
360 Everything a value can see or must refuse belongs to the file it is written
361 in — the values bound earlier in the same file, the names bound to an
362 encrypted value, the file's own path for error messages — so it lives on the
363 parser instead of being threaded through every step of parsing a value.
364 """
365
366 def __init__(
367 self,
368 content: str,
369 *,
370 source: Path | None = None,
371 skip_for: set[str] | None = None,
372 encrypted_names: set[str] | None = None,
373 expand: bool = True,
374 ) -> None:
375 """
376 Args:
377 content: The file's text
378 source: The file it was read from, for error messages
379 skip_for: Keys that are already bound, whose values are located but
380 never evaluated
381 encrypted_names: Names already bound to an encrypted value, which
382 can't be referenced from another value; encrypted keys met in
383 this file are added as we go
384 expand: If False, parse without expanding or running anything
385 """
386 self.content = content
387 self.source = source
388 self.skip_for = skip_for or set()
389 self.expand = expand
390 # Values bound earlier in this file, which later values can reference.
391 self.result: dict[str, str] = {}
392 self.encrypted_names = set(encrypted_names) if encrypted_names else set()
393
394 def parse(self, on_bind: Callable[[Binding], None]) -> dict[str, str]:
395 """Parse the whole file, calling `on_bind` with each binding, and return the key-value pairs."""
396 pos = 0
397 length = len(self.content)
398
399 while pos < length:
400 # Skip whitespace and empty lines
401 while pos < length and self.content[pos] in " \t\r\n":
402 pos += 1
403
404 if pos >= length:
405 break
406
407 # Skip comment lines
408 if self.content[pos] == "#":
409 pos = _skip_to_eol(self.content, pos)
410 continue
411
412 binding = self._binding(pos)
413 if binding is None:
414 # Skip to next line on parse failure
415 pos = _skip_to_eol(self.content, pos)
416 continue
417
418 self.result[binding.key] = binding.value
419 if binding.encrypted:
420 self.encrypted_names.add(binding.key)
421 on_bind(binding)
422 pos = binding.value_end
423
424 return self.result
425
426 def _binding(self, pos: int) -> Binding | None:
427 """Parse a KEY=value binding starting at `pos`, or return None if there isn't one."""
428 content = self.content
429 length = len(content)
430
431 # Skip optional 'export ' prefix
432 if content[pos : pos + 7] == "export ":
433 pos += 7
434 while pos < length and content[pos] in " \t":
435 pos += 1
436
437 # Parse key
438 key_start = pos
439 while pos < length and (content[pos].isalnum() or content[pos] == "_"):
440 pos += 1
441
442 if pos == key_start:
443 return None
444
445 key = content[key_start:pos]
446
447 # Must start with letter or underscore
448 if not (key[0].isalpha() or key[0] == "_"):
449 return None
450
451 # Skip whitespace before =
452 while pos < length and content[pos] in " \t":
453 pos += 1
454
455 # Expect =
456 if pos >= length or content[pos] != "=":
457 return None
458 pos += 1
459
460 # Skip whitespace after =
461 while pos < length and content[pos] in " \t":
462 pos += 1
463
464 # If the key is already bound, parse the value only to find where it ends
465 # (it may span lines when quoted) and keep the value already in place.
466 if key in self.skip_for:
467 _, value_end = self._value(pos, key, expand=False)
468 return Binding(
469 key=key,
470 value=os.environ.get(key, ""),
471 encrypted=False,
472 source=self.source,
473 key_start=key_start,
474 value_end=value_end,
475 )
476
477 # `encrypted:` at the start of an unquoted value is the encrypted form. It's
478 # recognized here, on the text as written, so the token is never expanded —
479 # and so a quoted value (which can't start with `e`) is always plain text.
480 if content.startswith(ENCRYPTED_VALUE_PREFIX, pos):
481 value, value_end = self._value(pos, key, expand=False)
482 return Binding(
483 key=key,
484 value=value,
485 encrypted=True,
486 source=self.source,
487 key_start=key_start,
488 value_end=value_end,
489 )
490
491 value, value_end = self._value(pos, key, expand=self.expand)
492 return Binding(
493 key=key,
494 value=value,
495 encrypted=False,
496 source=self.source,
497 key_start=key_start,
498 value_end=value_end,
499 )
500
501 def _value(self, pos: int, key: str, *, expand: bool) -> tuple[str, int]:
502 """Parse `key`'s value starting at pos, return (value, position just past it)."""
503 content = self.content
504
505 if pos >= len(content) or content[pos] in "\r\n":
506 return "", pos
507
508 char = content[pos]
509
510 # Single-quoted: literal value (no escape, no expansion), supports multiline
511 if char == "'":
512 return _parse_single_quoted(content, pos)
513
514 # Double-quoted: process escapes, variable expansion, and commands, supports multiline
515 if char == '"':
516 value, pos = _parse_double_quoted(content, pos)
517 if expand:
518 value = self._expand_variables(value, key)
519 value = _expand_commands(value)
520 value = value.replace(_ESCAPED_DOLLAR, "$") # Restore escaped $
521 return value, pos
522
523 # Unquoted value: variable expansion and command substitution
524 return self._unquoted(pos, key, expand=expand)
525
526 def _unquoted(self, pos: int, key: str, *, expand: bool) -> tuple[str, int]:
527 """Parse an unquoted value (until an inline comment or the end of the line)."""
528 content = self.content
529 start = pos
530 result = []
531 length = len(content)
532
533 while pos < length and content[pos] not in "\r\n":
534 char = content[pos]
535
536 # Stop at inline comment (whitespace followed by #)
537 if char == "#" and result and result[-1] in " \t":
538 break
539
540 # Handle backslash escapes (like bash)
541 if char == "\\" and pos + 1 < length:
542 next_char = content[pos + 1]
543 if next_char == "$":
544 result.append(_ESCAPED_DOLLAR) # Placeholder to prevent expansion
545 pos += 2
546 continue
547 elif next_char == "\\":
548 result.append("\\")
549 pos += 2
550 continue
551 # Other backslashes kept as-is
552
553 result.append(char)
554 pos += 1
555
556 # The value ends at its last non-whitespace character, so an inline comment
557 # or the line ending stays where it is.
558 while pos > start and content[pos - 1] in " \t":
559 pos -= 1
560
561 value = "".join(result).rstrip()
562
563 # Expand variables, then commands
564 if expand:
565 value = self._expand_variables(value, key)
566 value = _expand_commands(value)
567 value = value.replace(_ESCAPED_DOLLAR, "$") # Restore escaped $
568 return value, pos
569
570 def _expand_variables(self, value: str, key: str) -> str:
571 """Expand $VAR and ${VAR} references in `key`'s value.
572
573 Looks up variables in the values parsed so far first, then falls back to
574 os.environ. Unknown variables expand to an empty string. Referencing an
575 encrypted value is an error — it hasn't been decrypted yet, so expanding it
576 would either paste in the ciphertext or bind nothing at all.
577 """
578
579 def replacer(match: re.Match[str]) -> str:
580 var_name = match.group(1)
581 if var_name in self.encrypted_names:
582 raise ImproperlyConfigured(
583 f"{key} in {self.source} references {var_name}, which "
584 "is an encrypted value. Encrypted values can't be referenced "
585 "from other values."
586 )
587 # Check values defined earlier in this file, then os.environ
588 if var_name in self.result:
589 return self.result[var_name]
590 return os.environ.get(var_name, "")
591
592 # Expand ${VAR} first (more specific), then $VAR
593 value = _VAR_BRACE_RE.sub(replacer, value)
594 value = _VAR_BARE_RE.sub(replacer, value)
595 return value
596
597
598def _skip_to_eol(content: str, pos: int) -> int:
599 """Skip to end of line, return position after newline."""
600 while pos < len(content) and content[pos] not in "\r\n":
601 pos += 1
602 if pos < len(content) and content[pos] == "\r":
603 pos += 1
604 if pos < len(content) and content[pos] == "\n":
605 pos += 1
606 return pos
607
608
609def _parse_single_quoted(content: str, pos: int) -> tuple[str, int]:
610 """Parse single-quoted value (literal, multiline supported)."""
611 pos += 1 # Skip opening quote
612 start = pos
613 length = len(content)
614
615 while pos < length:
616 if content[pos] == "'":
617 value = content[start:pos]
618 return value, pos + 1
619 pos += 1
620
621 # No closing quote found, return what we have
622 return content[start:], pos
623
624
625def _parse_double_quoted(content: str, pos: int) -> tuple[str, int]:
626 """Parse double-quoted value (with escapes, multiline supported)."""
627 pos += 1 # Skip opening quote
628 result = []
629 length = len(content)
630
631 while pos < length:
632 char = content[pos]
633
634 if char == "\\" and pos + 1 < length:
635 next_char = content[pos + 1]
636 if next_char == "n":
637 result.append("\n")
638 elif next_char == "t":
639 result.append("\t")
640 elif next_char == "r":
641 result.append("\r")
642 elif next_char == '"':
643 result.append('"')
644 elif next_char == "\\":
645 result.append("\\")
646 elif next_char == "$":
647 result.append(_ESCAPED_DOLLAR) # Placeholder to prevent expansion
648 else:
649 # Unknown escape, keep both characters
650 result.append(char)
651 result.append(next_char)
652 pos += 2
653 elif char == '"':
654 return "".join(result), pos + 1
655 else:
656 result.append(char)
657 pos += 1
658
659 # No closing quote found, return what we have
660 return "".join(result), pos
661
662
663def _expand_commands(value: str) -> str:
664 """Expand all $(command) substitutions in value.
665
666 Handles nested parentheses within commands, e.g., $(echo "(test)").
667 """
668 result = []
669 i = 0
670 length = len(value)
671
672 while i < length:
673 # Look for $(
674 if i + 1 < length and value[i] == "$" and value[i + 1] == "(":
675 # Find matching closing paren, accounting for nesting
676 cmd_start = i + 2
677 depth = 1
678 j = cmd_start
679
680 while j < length and depth > 0:
681 if value[j] == "(":
682 depth += 1
683 elif value[j] == ")":
684 depth -= 1
685 j += 1
686
687 if depth == 0:
688 # Found matching ), extract and execute command
689 command = value[cmd_start : j - 1]
690 output = _execute_command(command)
691 result.append(output)
692 i = j
693 else:
694 # No matching ), keep literal
695 result.append(value[i])
696 i += 1
697 else:
698 result.append(value[i])
699 i += 1
700
701 return "".join(result)
702
703
704def _execute_command(command: str, timeout: float = 5.0) -> str:
705 """Execute a shell command and return stdout."""
706 try:
707 result = subprocess.run(
708 command,
709 shell=True,
710 stdout=subprocess.PIPE,
711 text=True,
712 timeout=timeout,
713 check=False,
714 )
715 return result.stdout.strip() if result.returncode == 0 else ""
716 except (subprocess.TimeoutExpired, OSError):
717 return ""