fix(issue-26): the wrapper reads .mam.env through the Python loader's parser, never source

- lib_py/delegation/envfile.py (stdlib only) holds the one set of rules:
  comments and blank lines skipped, key and value stripped, one matching
  quote pair removed, no shell interpretation, keys must match
  ^[A-Za-z_][A-Za-z0-9_]*$ (bad or `=`-less lines warn with the line
  number, never the value).
- mqtt_common._parse_env_file delegates to envfile.apply.
- The delegate-job wrapper runs envfile.py --shell by path and evals the
  shlex-quoted exports; values never reach a command line. The environment
  now wins over the file in both loaders.
- .mam.env.example: commented assignments no longer use $HOME, which the
  parser keeps literally.

The shell parsed the env file in one place only (wrapper :38), not three.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
GodopuandClaude Opus 5.5 committed 2026-09-30 15:47:40 +00:00
1 parent e1eb22bd59
commit f1331d36ce
5 files changed
+345 -19

No files matched your search

@@ -0,0 +1,78 @@
"""The one `.mam.env` parser (Issue #26).
Rules: blank lines and `#` comments are skipped; `KEY=VALUE` with the key and
the value stripped; one matching pair of surrounding quotes is removed; no
shell interpretation (`$(...)`, `${...}` and backticks stay literal). A key
must match ``^[A-Za-z_][A-Za-z0-9_]*$``; a line that does not, or that has no
`=`, is skipped with a warning naming the line (never the value — the file
holds secrets). Variables already in the environment win over the file.
Stdlib only: the delegate-job wrapper runs this file by path
(``envfile.py --shell <path>``) and evals the ``export`` lines it prints, so
values travel over a pipe, never on a command line.
"""
import os
import re
import shlex
import sys
from typing import List, MutableMapping, Tuple
_KEY = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
def parse(path: str) -> Tuple[List[Tuple[str, str]], List[str]]:
"""Return ([(key, value)], [warning]) for the file at ``path``."""
pairs: List[Tuple[str, str]] = []
warnings: List[str] = []
try:
with open(path, "r", encoding="utf-8") as f:
lines = f.read().splitlines()
except (OSError, UnicodeDecodeError) as e:
return pairs, [f"{path}: cannot read ({type(e).__name__})"]
for n, raw in enumerate(lines, 1):
line = raw.strip()
if not line or line.startswith("#"):
continue
key, sep, val = line.partition("=")
key = key.strip()
if not sep:
warnings.append(f"{path}:{n}: skipped (no '=')")
continue
if not _KEY.fullmatch(key):
warnings.append(f"{path}:{n}: skipped (invalid key)")
continue
val = val.strip()
if len(val) >= 2 and val[0] == val[-1] and val[0] in "\"'":
val = val[1:-1]
pairs.append((key, val))
return pairs, warnings
def apply(path: str, environ: MutableMapping[str, str]) -> Tuple[List[Tuple[str, str]], List[str]]:
"""Set each pair not already in ``environ``; return (the pairs set, warnings)."""
pairs, warnings = parse(path)
applied = []
for key, val in pairs:
if key not in environ:
environ[key] = val
applied.append((key, val))
return applied, warnings
def main(argv: List[str]) -> int:
if len(argv) != 2 or argv[0] != "--shell":
print("usage: envfile.py --shell <path>", file=sys.stderr)
return 2
path = argv[1]
if not os.path.isfile(path):
return 0
applied, warnings = apply(path, dict(os.environ))
for w in warnings:
print(f"WARNING: {w}", file=sys.stderr)
for key, val in applied:
print(f"export {key}={shlex.quote(val)}")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))
@@ -31,6 +31,8 @@ from typing import Any, Callable, Dict, Iterable, List, Optional
import paho.mqtt.client as mqtt
from lib_py.delegation import envfile
logger = logging.getLogger("delegate_job.mqtt_common")
_warned_deprecated_env = False
@@ -94,20 +96,9 @@ def _load_dotenv(workspace_dir: Optional[str] = None) -> None:
_parse_env_file(legacy_env_path)
def _parse_env_file(path: str) -> None:
try:
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" in line:
key, _, val = line.partition("=")
key = key.strip()
val = val.strip().strip('"').strip("'")
if key and key not in os.environ:
os.environ[key] = val
except Exception as e:
logger.warning("Failed to parse env file %s: %s", path, e)
"""Issue #26: the rules live in envfile.py, shared with the wrapper."""
for warning in envfile.apply(path, os.environ)[1]:
logger.warning("%s", warning)
_env_loaded = False
@@ -34,8 +34,12 @@ if [[ -z "$TARGET_ENV" ]]; then
fi
fi
# Issue #26: parsed by envfile.py (the Python loader's rules; environment wins),
# never sourced. Values come back as shlex-quoted exports over the pipe.
if [[ -n "$TARGET_ENV" && -f "$TARGET_ENV" ]]; then
set -a; source "$TARGET_ENV"; set +a
_env_exports="$("${DELEGATE_JOB_PYTHON:-python3}" -I "$SCRIPT_DIR/../lib_py/delegation/envfile.py" --shell "$TARGET_ENV")" \
|| { echo "ERROR: cannot read $TARGET_ENV" >&2; exit 1; }
eval "$_env_exports"; unset _env_exports
fi
# Source EARLY (before any herdr usage in run_agent) — this is what turns