1
"""Keep credential migration, exports and deployments consistent using fixtures."""
2
import importlib.util
3
import json
4
import os
5
from pathlib import Path
6
import subprocess
7
import tempfile
8
from types import SimpleNamespace
9
import unittest
10
from unittest.mock import patch
11
 
12
from config import env_value, read_env
13
from credentials import (
14
    EXPORT_HEADER,
15
    WORKER_KEYS,
16
    initialize_credentials,
17
    sync_worker_secrets,
18
    worker_secrets,
19
)
20
 
21
spec = importlib.util.spec_from_file_location(
22
    "deploy_worker", Path(__file__).with_name("deploy_worker.py")
23
)
24
deploy_worker = importlib.util.module_from_spec(spec)
25
spec.loader.exec_module(deploy_worker)
26
 
27
BASE = {
28
    "REALTIME_APP_ID": "synthetic-app-id",
29
    "REALTIME_APP_TOKEN": "synthetic-app-token",
30
    "WIFI_SSID": "Synthetic network",
31
    "WIFI_PASSWORD": "synthetic-wifi-password",
32
}
33
TOKENS = {"DEVICE_TOKEN": "d" * 43, "VIEWER_PASSWORD": "synthetic-viewer-password"}
34
 
35
 
36
def write_env(path, values):
37
    path.write_text("".join(f"{key}={env_value(key, value)}\n" for key, value in values.items()))
38
 
39
 
40
class CredentialTests(unittest.TestCase):
41
    def setUp(self):
42
        self.directory = tempfile.TemporaryDirectory()
43
        self.addCleanup(self.directory.cleanup)
44
        self.root = Path(self.directory.name)
45
        (self.root / "worker").mkdir()
46
        self.source = self.root / ".credential.env"
47
        self.export = self.root / "worker/.dev.vars"
48
 
49
    def test_migration_preserves_credentials_and_source_comments(self):
50
        write_env(self.source, BASE)
51
        original = "# Keep this private.\n" + self.source.read_text()
52
        self.source.write_text(original)
53
        self.export.write_text("".join(
54
            f"{key}={json.dumps(value)}\n"
55
            for key, value in {**BASE, **TOKENS, "LOCAL_DEMO_ORIGIN": "obsolete"}.items()
56
        ))
57
        with patch("credentials.secrets.token_urlsafe") as generate:
58
            initialize_credentials(self.root)
59
        generate.assert_not_called()
60
        self.assertEqual(read_env(self.source), {**BASE, **TOKENS})
61
        self.assertTrue(self.source.read_text().startswith(original))
62
        self.assertEqual(read_env(self.export), {key: {**BASE, **TOKENS}[key] for key in WORKER_KEYS})
63
        self.assertTrue(self.export.read_text().startswith(EXPORT_HEADER))
64
        self.assertEqual(self.source.stat().st_mode & 0o777, 0o600)
65
        self.assertEqual(self.export.stat().st_mode & 0o777, 0o600)
66
        original_source = self.source.read_bytes()
67
        export_time = self.export.stat().st_mtime_ns
68
        with patch("credentials.secrets.token_urlsafe") as generate:
69
            initialize_credentials(self.root)
70
        generate.assert_not_called()
71
        self.assertEqual(self.source.read_bytes(), original_source)
72
        self.assertEqual(self.export.stat().st_mtime_ns, export_time)
73
 
74
    def test_first_setup_fills_empty_template_fields_once(self):
75
        write_env(self.source, {**BASE, "DEVICE_TOKEN": "", "VIEWER_PASSWORD": ""})
76
        with patch("credentials.secrets.token_urlsafe", side_effect=TOKENS.values()) as generate:
77
            initialize_credentials(self.root)
78
        self.assertEqual(generate.call_count, 2)
79
        self.assertEqual(read_env(self.source), {**BASE, **TOKENS})
80
        with patch("credentials.secrets.token_urlsafe") as generate:
81
            initialize_credentials(self.root)
82
        generate.assert_not_called()
83
 
84
    def test_canonical_values_override_stale_exports_and_can_recreate_them(self):
85
        write_env(self.source, {**BASE, **TOKENS})
86
        initialize_credentials(self.root)
87
        changed = {**BASE, **TOKENS, "VIEWER_PASSWORD": "changed # with spaces"}
88
        write_env(self.source, changed)
89
        self.assertEqual(worker_secrets(self.root)["VIEWER_PASSWORD"], changed["VIEWER_PASSWORD"])
90
        sync_worker_secrets(self.root)
91
        self.assertEqual(read_env(self.export), {key: changed[key] for key in WORKER_KEYS})
92
        self.export.unlink()
93
        sync_worker_secrets(self.root)
94
        self.assertEqual(read_env(self.export), {key: changed[key] for key in WORKER_KEYS})
95
 
96
    def test_explicit_canonical_token_wins_during_migration(self):
97
        write_env(self.source, {**BASE, "DEVICE_TOKEN": TOKENS["DEVICE_TOKEN"]})
98
        write_env(self.export, {**TOKENS, "DEVICE_TOKEN": "obsolete" * 6})
99
        initialize_credentials(self.root)
100
        self.assertEqual(read_env(self.source), {**BASE, **TOKENS})
101
 
102
    def test_routine_sync_never_generates_missing_credentials(self):
103
        write_env(self.source, BASE)
104
        original = self.source.read_bytes()
105
        with patch("credentials.secrets.token_urlsafe") as generate:
106
            with self.assertRaisesRegex(SystemExit, "DEVICE_TOKEN, VIEWER_PASSWORD"):
107
                sync_worker_secrets(self.root)
108
        generate.assert_not_called()
109
        self.assertEqual(self.source.read_bytes(), original)
110
        self.assertFalse(self.export.exists())
111
 
112
    def test_initialized_credentials_are_not_recovered_from_generated_export(self):
113
        write_env(self.source, {**BASE, **TOKENS})
114
        initialize_credentials(self.root)
115
        write_env(self.source, {**BASE, "DEVICE_TOKEN": TOKENS["DEVICE_TOKEN"]})
116
        original_export = self.export.read_bytes()
117
        with patch("credentials.secrets.token_urlsafe") as generate:
118
            with self.assertRaisesRegex(SystemExit, "Restore VIEWER_PASSWORD"):
119
                initialize_credentials(self.root)
120
        generate.assert_not_called()
121
        self.assertEqual(self.export.read_bytes(), original_export)
122
 
123
    def test_incomplete_legacy_export_is_not_replaced(self):
124
        write_env(self.source, BASE)
125
        write_env(self.export, {"VIEWER_PASSWORD": TOKENS["VIEWER_PASSWORD"]})
126
        original_source, original_export = self.source.read_bytes(), self.export.read_bytes()
127
        with self.assertRaisesRegex(SystemExit, "Legacy worker/.dev.vars is incomplete"):
128
            initialize_credentials(self.root)
129
        self.assertEqual(self.source.read_bytes(), original_source)
130
        self.assertEqual(self.export.read_bytes(), original_export)
131
 
132
    def test_frontend_checkout_can_skip_secrets_but_never_use_an_orphan_export(self):
133
        sync_worker_secrets(self.root, optional=True)
134
        self.assertFalse(self.source.exists())
135
        self.assertFalse(self.export.exists())
136
        write_env(self.export, TOKENS)
137
        with self.assertRaisesRegex(SystemExit, "Create .credential.env"):
138
            sync_worker_secrets(self.root, optional=True)
139
 
140
    def test_invalid_credentials_do_not_modify_private_files(self):
141
        for overrides in ({"DEVICE_TOKEN": "short"}, {"VIEWER_PASSWORD": "x" * 257}):
142
            with self.subTest(keys=tuple(overrides)):
143
                write_env(self.source, {**BASE, **TOKENS, **overrides})
144
                original = self.source.read_bytes()
145
                with self.assertRaises(SystemExit):
146
                    initialize_credentials(self.root)
147
                self.assertEqual(self.source.read_bytes(), original)
148
                self.assertFalse(self.export.exists())
149
 
150
    def test_deploy_uploads_current_canonical_worker_secrets_only(self):
151
        write_env(self.source, {**BASE, **TOKENS})
152
        initialize_credentials(self.root)
153
        changed = {**BASE, **TOKENS, "VIEWER_PASSWORD": "new canonical password"}
154
        write_env(self.source, changed)
155
        config = self.root / "worker/dist/esp32_radio/wrangler.json"
156
        config.parent.mkdir(parents=True)
157
        config.write_text("{}")
158
        uploaded = []
159
 
160
        def run(command, **kwargs):
161
            path = Path(command[command.index("--secrets-file") + 1])
162
            uploaded.append(json.loads(path.read_text()))
163
            self.assertEqual(path.stat().st_mode & 0o777, 0o600)
164
            return SimpleNamespace(returncode=0)
165
 
166
        original_umask = os.umask(0o077)
167
        self.addCleanup(os.umask, original_umask)
168
        with patch.object(deploy_worker, "ROOT", self.root), patch.object(deploy_worker.subprocess, "run", side_effect=run):
169
            with self.assertRaises(SystemExit) as result:
170
                deploy_worker.main([])
171
        self.assertEqual(result.exception.code, 0)
172
        self.assertEqual(uploaded, [{key: changed[key] for key in WORKER_KEYS}])
173
 
174
    def test_python_node_and_wrangler_agree_on_literal_credentials(self):
175
        values = {
176
            "SPACES": "  two  spaces  ",
177
            "PUNCTUATION": "# $HOME $(command) `text` \\",
178
            "SINGLE_QUOTE": "it's private",
179
            "DOUBLE_QUOTE": 'say "hello"',
180
            "BOTH_QUOTES": "it's a \"password\"",
181
            "BACKSLASH": r"keep\n\r\t literal",
182
            "UNICODE": "café ☕",
183
            "EMPTY": "",
184
        }
185
        write_env(self.source, values)
186
        self.assertEqual(read_env(self.source), values)
187
        script = "import {parseEnv} from 'node:util'; import {readFileSync} from 'node:fs'; process.stdout.write(JSON.stringify(parseEnv(readFileSync(process.argv[1], 'utf8'))));"
188
        decoded = json.loads(subprocess.check_output(
189
            ["node", "--input-type=module", "-e", script, str(self.source)], text=True,
190
        ))
191
        self.assertEqual(decoded, values)
192
        self.export.write_text(self.source.read_text())
193
        script = "import {unstable_getVarsForDev as read} from 'wrangler'; const values = read(process.argv[1], undefined, {}, undefined, true); process.stdout.write(JSON.stringify(Object.fromEntries(Object.entries(values).map(([key, binding]) => [key, binding.value]))));"
194
        decoded = json.loads(subprocess.check_output(
195
            ["node", "--input-type=module", "-e", script, str(self.root / "worker/wrangler.jsonc")],
196
            cwd=Path(__file__).resolve().parents[1] / "worker", text=True,
197
        ))
198
        self.assertEqual(decoded, values)
199
 
200
    def test_dotenv_comments_export_and_quotes_are_literal(self):
201
        self.source.write_text(
202
            "# comment\nexport UNQUOTED = two  spaces # comment\n"
203
            "QUOTED=' # $HOME `command` ' # comment\n"
204
        )
205
        self.assertEqual(read_env(self.source), {
206
            "UNQUOTED": "two  spaces", "QUOTED": " # $HOME `command` ",
207
        })
208
 
209
    def test_parse_errors_do_not_include_secret_values(self):
210
        for contents in (
211
            "not-an-assignment synthetic-secret",
212
            "VALUE='synthetic-secret",
213
            "VALUE='synthetic-secret' extra",
214
            "VALUE='synthetic-secret'\nVALUE=duplicate",
215
        ):
216
            self.source.write_text(contents)
217
            with self.assertRaises(SystemExit) as error:
218
                read_env(self.source)
219
            self.assertNotIn("synthetic-secret", str(error.exception))
220
 
221
 
222
if __name__ == "__main__":
223
    unittest.main()