1
"""Canonical private credentials and the generated Wrangler development export."""
2
import os
3
import secrets
4
import tempfile
5
 
6
from config import ENV_ASSIGNMENT, env_value, read_env
7
 
8
WORKER_KEYS = ("REALTIME_APP_ID", "REALTIME_APP_TOKEN", "DEVICE_TOKEN", "VIEWER_PASSWORD")
9
GENERATED_KEYS = ("DEVICE_TOKEN", "VIEWER_PASSWORD")
10
EXPORT_HEADER = "# Generated from ../.credential.env. Edit that file; this export is overwritten.\n"
11
 
12
 
13
def private_write(path, contents):
14
    """Replace a private file atomically, keeping unchanged exports stable."""
15
    if path.exists() and path.read_text() == contents:
16
        path.chmod(0o600)
17
        return
18
    descriptor, name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
19
    try:
20
        with os.fdopen(descriptor, "w") as output:
21
            output.write(contents)
22
        os.replace(name, path)
23
    finally:
24
        if os.path.exists(name):
25
            os.unlink(name)
26
 
27
 
28
def validate_worker(values):
29
    missing = [key for key in WORKER_KEYS if not values.get(key)]
30
    if missing:
31
        raise SystemExit(
32
            f"Set {', '.join(missing)} in .credential.env; run make secrets for initial setup."
33
        )
34
    if not 32 <= len(values["DEVICE_TOKEN"]) <= 128:
35
        raise SystemExit("DEVICE_TOKEN in .credential.env must be 32–128 characters")
36
    if len(values["VIEWER_PASSWORD"].encode("utf-16-le")) // 2 > 256:
37
        raise SystemExit("VIEWER_PASSWORD in .credential.env must be at most 256 characters")
38
    return {key: values[key] for key in WORKER_KEYS}
39
 
40
 
41
def worker_secrets(root):
42
    path = root / ".credential.env"
43
    if not path.exists():
44
        raise SystemExit("Create .credential.env from .credential.env.example and run make secrets")
45
    return validate_worker(read_env(path))
46
 
47
 
48
def export_contents(values):
49
    return EXPORT_HEADER + "".join(f"{key}={env_value(key, values[key])}\n" for key in WORKER_KEYS)
50
 
51
 
52
def sync_worker_secrets(root, *, optional=False):
53
    """Refresh only the export; ordinary commands never create credentials."""
54
    source = root / ".credential.env"
55
    export = root / "worker/.dev.vars"
56
    if optional and not source.exists() and not export.exists():
57
        return
58
    values = worker_secrets(root)
59
    private_write(export, export_contents(values))
60
 
61
 
62
def initialize_credentials(root):
63
    """Import credentials from an older export, or generate them during first setup."""
64
    source = root / ".credential.env"
65
    export = root / "worker/.dev.vars"
66
    if not source.exists():
67
        raise SystemExit("Create .credential.env from .credential.env.example first")
68
    contents = source.read_text()
69
    values = read_env(source)
70
    missing = [key for key in GENERATED_KEYS if not values.get(key)]
71
    updates = {}
72
    if missing and export.exists():
73
        if export.read_text().startswith(EXPORT_HEADER):
74
            raise SystemExit(
75
                f"Restore {', '.join(missing)} in .credential.env; credentials are already initialized."
76
            )
77
        legacy = read_env(export)
78
        if any(not legacy.get(key) for key in missing):
79
            raise SystemExit("Legacy worker/.dev.vars is incomplete; restore its device/viewer credentials first")
80
        updates = {key: legacy[key] for key in missing}
81
    elif missing:
82
        updates = {key: secrets.token_urlsafe(32) for key in missing}
83
    values.update(updates)
84
    rendered = export_contents(validate_worker(values))
85
    remaining = updates.copy()
86
    lines = []
87
    for line in contents.splitlines(keepends=True):
88
        match = ENV_ASSIGNMENT.fullmatch(line.rstrip("\r\n"))
89
        if match and match[1] in remaining:
90
            key = match[1]
91
            lines.append(f"{key}={env_value(key, remaining.pop(key))}\n")
92
        else:
93
            lines.append(line)
94
    contents = "".join(lines)
95
    if remaining:
96
        if contents and not contents.endswith("\n"):
97
            contents += "\n"
98
        contents += "".join(f"{key}={env_value(key, value)}\n" for key, value in remaining.items())
99
    private_write(source, contents)
100
    private_write(export, rendered)