#!/usr/bin/env python3 """Synthetic regressions for private runtime-env transport and Compose quoting.""" from __future__ import annotations import json import os import shutil import stat import subprocess import sys import tempfile import unittest from pathlib import Path import runtime_env_parse as parser SCRIPT = Path(__file__).with_name("runtime_env.py") PEM = "-----BEGIN PRIVATE KEY-----\nU1lOVEhFVElDLU5PVC1BLUtFWQ==\n-----END PRIVATE KEY-----" JSON_VALUE = json.dumps( { "type": "service_account", "private_key": PEM + "\n", "literal": "$ENV_SENTINEL ${ENV_SENTINEL} $$ # retained", "apostrophe": r"C:\\can't", "unicode": "café ∑", "nested": [{"enabled": True, "missing": None}], }, indent=2, ensure_ascii=False, ) SENTINEL = "SYNTHETIC_VALUE_MUST_NOT_APPEAR_IN_ERRORS" def run_cli(*args: str, value: str = "", payload: bytes | None = None): environment = { "PATH": os.environ.get("PATH", os.defpath), "PYTHONIOENCODING": "utf-8", "RUNTIME_ENV": value, } return subprocess.run( [sys.executable, str(SCRIPT), *args], input=payload, env=environment, capture_output=True, timeout=10, check=False, ) def compose_command(): standalone = os.environ.get("DOCKER_COMPOSE_BIN") or shutil.which("docker-compose") command = [standalone] if standalone else ["docker", "compose"] try: available = subprocess.run(command + ["version", "--short"], capture_output=True, timeout=10) except (OSError, subprocess.TimeoutExpired): available = None if available is None or available.returncode: if os.environ.get("GOWALK_REQUIRE_COMPOSE_TEST") == "1": raise AssertionError("Compose CLI is required for the runtime-env compatibility test") raise unittest.SkipTest("Compose CLI is unavailable; CI requires the real-parser check") return command class RuntimeEnvTest(unittest.TestCase): def test_existing_dotenv_is_byte_preserved(self): samples = [ "", "# comment\nEMPTY=\nFOO=bar # comment\n", "ONE=1\r\nTWO=2\r\n", "export lower.case-name = ${BASE:-fallback}\nINHERITED\n", "lower: value\n", "JSON={ordinary-single-line-value\n", 'JSON={"price":"$BASE","literal":"\\\\n"}\n', "ONE='literal $BASE \\n'\nTWO=\"${ONE}-suffix\"\n", "JSON='{\n \"price\": \"$BASE\"\n}' # already quoted\nNEXT=yes\n", 'MULTI="first\nsecond"\nAFTER=ok', "UNICODE\u00a0=\u00a0'literal\nPEM=-----BEGIN PRIVATE KEY-----\n'\nAFTER=ok\n", "café[0]='literal\nJSON={\nnot structured here\n'\n", "JSON={ordinary-single-line-value\nAFTER=yes\n", "\ufeffONE=valid\n", "MULTI='first\nPEM=-----BEGIN PRIVATE KEY-----\nJSON={\nlast'\nAFTER=ok\n", ] for value in samples: with self.subTest(sample=samples.index(value)): self.assertEqual(parser.normalize(value), value) def test_raw_multiline_structures_are_quoted_without_expansion(self): source = "BEFORE=${BASE:-old}\nPEM=" + PEM + "\nJSON=" + JSON_VALUE + "\nAFTER=yes\n" normalized = parser.normalize(source) self.assertTrue(normalized.startswith("BEFORE=${BASE:-old}\nPEM='-----BEGIN PRIVATE KEY-----\n")) self.assertIn("\nJSON='" + JSON_VALUE.replace("'", "\\'") + "'\n", normalized) self.assertTrue(normalized.endswith("\nAFTER=yes\n")) self.assertEqual(parser.normalize(normalized), normalized) array = '[\n {"value": "$BASE", "name": "O\'Brien"}\n]' self.assertEqual(parser.normalize("ARRAY=" + array + "\n"), "ARRAY='" + array.replace("'", "\\'") + "'\n") def test_malformed_structures_fail_without_values_in_errors(self): cases = [ "PEM=-----BEGIN PRIVATE KEY-----\n" + SENTINEL + "\n", "PEM=" + PEM.replace("END PRIVATE KEY", "END CERTIFICATE") + "\n", "PEM=" + PEM.replace("U1lOVEhFVElDLU5PVC1BLUtFWQ==", "NEXT=owned") + "\n", "PEM=" + PEM + " trailing-material\n", "PEM=" + PEM + "\n" + PEM + "\n", 'JSON={\n"secret": "' + SENTINEL + '"\n', 'JSON={\n"value": 1\n} trailing-material\n', 'JSON={\n"same": 1, "same": 2\n}\n', 'JSON={\n"value": NaN\n}\n', 'JSON={\n"value": Infinity\n}\n', 'JSON={\n"value": "bad\\escape"\n}\n', 'JSON={\n"value": "unterminated\nNEXT=owned\n}\n', "BAD%=\"" + SENTINEL + "\"\n", "BAD/NAME=" + SENTINEL + "\n", "VALUE='safe' malformed-" + SENTINEL + "\n", "EXISTING='" + SENTINEL + "\n", 'EXISTING="' + SENTINEL + "\n", "JSON=[\n" + "[" * 1100 + "0" + "]" * 1100 + "\n]\n", ] for index, value in enumerate(cases): with self.subTest(case=index): with self.assertRaises(parser.RuntimeEnvError) as caught: parser.normalize(value) self.assertRegex(str(caught.exception), r"^runtime_env_[a-z_]+$") self.assertNotIn(SENTINEL, str(caught.exception)) def test_input_bounds_and_invalid_text_fail_closed(self): for value in ["A=" + "x" * (2 * 1024 * 1024), "A=" + "∑" * (1024 * 1024), "A=x\x00y", "A=\ud800"]: with self.subTest(length=len(value)): with self.assertRaises(parser.RuntimeEnvError): parser.normalize(value) def test_private_transport_roundtrip_and_atomic_replacement(self): values = ["", "FOO=${BASE:-original}\n", "PEM=" + PEM + "\nJSON=" + JSON_VALUE + "\n"] with tempfile.TemporaryDirectory() as directory: target = Path(directory, ".runtime.env") for index, value in enumerate(values): with self.subTest(case=index): target.write_text("OLD=unchanged\n", encoding="utf-8") target.chmod(0o644) exported = run_cli("export", value=value) self.assertEqual(exported.returncode, 0, exported.stderr.decode()) received = run_cli("receive", str(target), payload=exported.stdout) self.assertEqual(received.returncode, 0, received.stderr.decode()) self.assertEqual(received.stdout, b"") self.assertEqual(received.stderr, b"") self.assertEqual(target.read_text(), parser.normalize(value)) self.assertEqual(stat.S_IMODE(target.stat().st_mode), 0o600) self.assertEqual(list(Path(directory).iterdir()), [target]) original = Path(directory, "original") original.write_text("DO_NOT_FOLLOW", encoding="utf-8") target.unlink() target.symlink_to(original) received = run_cli("receive", str(target), payload=run_cli("export", value="NEW=yes\n").stdout) self.assertEqual(received.returncode, 0, received.stderr.decode()) self.assertEqual(original.read_text(), "DO_NOT_FOLLOW") self.assertFalse(target.is_symlink()) self.assertEqual(target.read_text(), "NEW=yes\n") def test_failed_receive_preserves_target_and_redacts_values(self): malformed = 'JSON={\n"secret": "' + SENTINEL + '"\n' exported = run_cli("export", value=malformed) with tempfile.TemporaryDirectory() as directory: target = Path(directory, ".runtime.env") target.write_text("OLD=unchanged\n", encoding="utf-8") payloads = [b"\xff", b"not-json", b"{}", b" " * (2 * 1024 * 1024)] if exported.returncode == 0: payloads.append(exported.stdout) else: self.assertNotIn(SENTINEL.encode(), exported.stdout + exported.stderr) for index, payload in enumerate(payloads): with self.subTest(case=index): received = run_cli("receive", str(target), payload=payload) self.assertNotEqual(received.returncode, 0) self.assertEqual(received.stdout, b"") self.assertNotIn(SENTINEL.encode(), received.stderr) self.assertNotIn(b"Traceback", received.stderr) self.assertEqual(target.read_text(), "OLD=unchanged\n") self.assertEqual(list(Path(directory).iterdir()), [target]) target.unlink() target.mkdir() received = run_cli("receive", str(target), payload=run_cli("export", value="NEW=yes\n").stdout) self.assertNotEqual(received.returncode, 0) self.assertEqual(list(target.iterdir()), []) self.assertEqual(list(Path(directory).iterdir()), [target]) self.assertNotIn(b"Traceback", received.stderr) def test_real_compose_parser_preserves_credential_and_legacy_values(self): command = compose_command() value = "PEM=" + PEM + "\nJSON=" + JSON_VALUE + "\nBASE=legacy\nLEGACY=${BASE}-suffix\n" with tempfile.TemporaryDirectory() as directory: root = Path(directory) runtime = root / ".runtime.env" runtime.write_text(parser.normalize(value), encoding="utf-8") compose = root / "compose.yaml" compose.write_text( "services:\n fixture:\n image: scratch\n environment:\n" " PEM: ${PEM}\n JSON: ${JSON}\n LEGACY: ${LEGACY}\n", encoding="utf-8", ) result = subprocess.run( command + [ "--env-file", str(runtime), "-f", str(compose), "-p", "runtime-env-test", "config", "--format", "json", ], capture_output=True, timeout=20, check=False, env={"PATH": os.environ.get("PATH", os.defpath), "ENV_SENTINEL": "MUST_NOT_EXPAND"}, ) self.assertEqual(result.returncode, 0, result.stderr.decode()) actual = json.loads(result.stdout)["services"]["fixture"]["environment"] self.assertEqual(actual["PEM"].rstrip("\n"), PEM) # Canonical config re-escapes dollars so the output can be parsed again safely. self.assertEqual(actual["JSON"], JSON_VALUE.replace("$", "$$")) self.assertEqual(actual["LEGACY"], "legacy-suffix") environment = subprocess.run( command + ["--env-file", str(runtime), "-f", str(compose), "config", "--environment"], capture_output=True, timeout=20, check=False, env={"PATH": os.environ.get("PATH", os.defpath), "ENV_SENTINEL": "MUST_NOT_EXPAND"}, ) self.assertEqual(environment.returncode, 0, environment.stderr.decode()) resolved = "\n" + environment.stdout.decode() self.assertIn("\nJSON=" + JSON_VALUE + "\n", resolved) self.assertIn("\nPEM=" + PEM + "\n", resolved) self.assertIn("\nLEGACY=legacy-suffix\n", resolved) if __name__ == "__main__": unittest.main()