feat: the .env layer the management page will write through

envfile.py edits the file in place, line by line: only the KEY= line of a
key that actually changed is replaced, so the paragraph explaining why
GROUP_IDS is empty survives an edit made in a browser. A regenerated file
would be correct and would still lose most of what that file is for.
Temp-then-replace with the mode reapplied before the swap, because .env
holds the account number and is edited from a tab that can be closed
mid-save.

envkeys.py is the policy, as data. Nothing here re-implements validation:
load_config(candidate, use_dotenv=False) is what startup runs, so an edit
this bot would refuse at boot is refused at the form instead, with the
journal's own message.

Read-only is wider than "the phone number". API_BASE_URL repoints every
outgoing message and attachment fetch at the next restart, which is worse
than changing the number; DATABASE_PATH orphans the history; PHOTO_PACK_DIR
cannot be validated at all, because _photo_pack_dirs() reads os.environ
directly rather than the mapping it is handed -- deliberately, since that is
what keeps the suite away from a real album.

GROUP_IDS and SELFTEST_GROUPS are editable but refused while no group
listing has been read. parse_group_ids only checks syntax; startup also
requires the group to be one the bot actually joined, so writing one blind
with the container down turns a scoping mistake into a unit that will not
start.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01416TegJ3sa8xL5qGGgrygP
This commit is contained in:
Krzysztof Rudnicki 2026-08-31 14:31:29 +02:00
parent 3a348a550c
commit bdd20654bf
5 changed files with 415 additions and 0 deletions

View File

@ -0,0 +1,2 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The management page: its routes, and the files they read and write."""

View File

@ -0,0 +1,123 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Rewriting ``.env`` without losing what it says about itself.
The file is edited in place, line by line: only the ``KEY=`` line of a key
that actually changed is replaced, so every comment in ``.env.sample``'s
descendant survives an edit made in a browser. A regenerated file would be
correct and would still lose the paragraph explaining why ``GROUP_IDS`` is
empty, which is most of what that file is.
Nothing here validates. ``admin_config`` builds the candidate mapping and
hands it to :func:`signal_bot.config_env.load_config` -- the same function
startup uses, so an edit this repository would refuse at boot is refused at
the form instead of eight hours later in ``systemctl --user --failed``.
Which keys may be edited at all is policy, and lives in ``envkeys``. This
module only knows how to read the file and put it back.
"""
from __future__ import annotations
import os
from typing import TYPE_CHECKING, Final
if TYPE_CHECKING:
from collections.abc import Iterator, Mapping
from pathlib import Path
# .env is a secret: it holds the account's number and the identities allowed
# to add the bot to a group. Rewriting must not widen it.
FILE_MODE: Final = 0o600
_MARKER: Final = "# Added by the management page."
def parse(text: str) -> dict[str, str]:
"""Every ``KEY=value`` in a ``.env``, comments and blanks ignored.
Args:
text: The file's contents.
Returns:
The keys in file order, values stripped of surrounding whitespace.
"""
return {key: value for key, value, _ in _assignments(text)}
def _assignments(text: str) -> Iterator[tuple[str, str, int]]:
"""Yield ``(key, value, line index)`` for every assignment line."""
for index, line in enumerate(text.splitlines()):
stripped = line.strip()
if not stripped or stripped.startswith("#") or "=" not in stripped:
continue
key, _, value = stripped.partition("=")
key = key.strip()
if key:
yield key, value.strip(), index
def candidate(text: str, changes: Mapping[str, str]) -> dict[str, str]:
"""The environment the bot would load if ``changes`` were written.
The process environment is included underneath because that is the order
``load_config`` sees in production: ``load_dotenv`` does not overwrite a
value systemd already exported, so a check that ignored it would validate
a file the running unit does not actually use.
Args:
text: The current ``.env`` contents.
changes: The edited keys and their new values.
Returns:
A mapping suitable for ``load_config(env, use_dotenv=False)``.
"""
merged = dict(parse(text))
merged.update(changes)
for key, value in os.environ.items():
if key not in changes:
merged.setdefault(key, value)
return merged
def render(text: str, changes: Mapping[str, str]) -> str:
"""The rewritten file: same comments, same order, new values.
A key already present is replaced where it stands. A key absent from the
file is appended under a marker rather than dropped into the middle,
because the surrounding comments explain their own key and not this one.
Args:
text: The current ``.env`` contents.
changes: The edited keys and their new values.
Returns:
The full new contents, newline-terminated.
"""
lines = text.splitlines()
seen: set[str] = set()
for key, _, index in _assignments(text):
if key in changes:
lines[index] = f"{key}={changes[key]}"
seen.add(key)
appended = [key for key in changes if key not in seen]
if appended:
lines.extend(["", _MARKER, *(f"{key}={changes[key]}" for key in appended)])
return "\n".join(lines) + "\n"
def write(path: Path, changes: Mapping[str, str]) -> None:
"""Replace ``.env`` atomically, keeping its mode.
Temp-then-replace for the same reason the album's manifests use it: a
half-written ``.env`` is not a bad value, it is a bot that will not start,
and this file is edited from a browser tab that can be closed mid-save.
Args:
path: The ``.env`` to rewrite.
changes: The edited keys and their new values.
"""
text = path.read_text(encoding="utf-8")
temporary = path.with_name(path.name + ".tmp")
temporary.write_text(render(text, changes), encoding="utf-8")
temporary.chmod(FILE_MODE)
temporary.replace(path)

View File

@ -0,0 +1,106 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Which ``.env`` keys a browser may write, and what makes an edit valid.
The tables are the whole policy and they are data, so "can the page change
this?" has one answer and it is greppable. Validation is deliberately not
re-implemented here: :func:`signal_bot.config_env.load_config` is what
startup runs, so handing it the candidate mapping means an edit this bot
would refuse at boot is refused at the form instead -- and an edit it accepts
here is one it will accept there.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Final
from signal_bot.config_env import load_config
from signal_bot.manage import envfile
if TYPE_CHECKING:
from collections.abc import Mapping
# Everything the page may write. The read-only remainder is still shown --
# "what is this bot configured with" is the first question the page answers --
# but PHONE_NUMBER, DATABASE_PATH, PHOTO_PACK_DIR and the bind address all
# name things outside the file, and changing them from a form would leave the
# container, the database or the album pointing somewhere that does not exist.
EDITABLE: Final = (
"GROUP_IDS",
"TRUSTED_INVITERS",
"SELFTEST_GROUPS",
"COOLDOWN_SECONDS",
"REPORT_HOUR",
"REPORT_MINUTE",
"SILENCE_SECONDS",
"LOG_LEVEL",
)
# Shown, never written.
READ_ONLY: Final = (
"PHONE_NUMBER",
"API_BASE_URL",
"DATABASE_PATH",
"PHOTO_PACK_DIR",
"BIND_HOST",
"BIND_PORT",
)
# Two of the editable keys name groups, and a group id is only meaningful
# against a listing that was actually read. With the container down the
# picker has nothing to offer, and a free-text fallback would let a
# syntactically valid write name a group the next startup rejects -- turning
# a scoping mistake into a failed unit. So these are refused in that state.
NEEDS_GROUPS: Final = ("GROUP_IDS", "SELFTEST_GROUPS")
def settings(text: str) -> list[dict[str, str | bool]]:
"""Every key the page shows, with whether it may be edited.
Args:
text: The current ``.env`` contents.
Returns:
One entry per known key, editable ones first, each carrying the
current value -- empty when the key is absent, which is how ``.env``
spells "use the default".
"""
values = envfile.parse(text)
return [
{"key": key, "value": values.get(key, ""), "editable": key in EDITABLE}
for key in (*EDITABLE, *READ_ONLY)
]
def refuse(changes: Mapping[str, str], *, groups_known: bool) -> str | None:
"""The first key that may not be written right now, or None.
Args:
changes: The edited keys and their new values.
groups_known: Whether a group listing was actually read. With the
container down it was not, and a group id checked only for syntax
can name a group the next startup rejects -- which turns a scoping
mistake into a unit that will not start.
Returns:
The offending key, or None when every change is allowed.
"""
for key in changes:
if key not in EDITABLE:
return key
if key in NEEDS_GROUPS and not groups_known:
return key
return None
def validate(text: str, changes: Mapping[str, str]) -> None:
"""Prove the bot could start on the edited file.
Args:
text: The current ``.env`` contents.
changes: The edited keys and their new values.
Raises:
ConfigError: The result is something ``load_config`` refuses. The
message is the one startup would print, verbatim.
"""
load_config(envfile.candidate(text, changes), use_dotenv=False)

View File

@ -0,0 +1,98 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Rewriting .env without losing what it says about itself."""
from __future__ import annotations
from typing import TYPE_CHECKING
from signal_bot.manage import envfile
if TYPE_CHECKING:
from pathlib import Path
import pytest
SAMPLE = """\
# The account, in +E.164 form.
PHONE_NUMBER=+15550000000
# Empty means every group this account is in.
GROUP_IDS=
COOLDOWN_SECONDS=10
"""
def test_comments_blank_lines_and_junk_are_not_assignments() -> None:
text = "# a comment\n\nNOT_AN_ASSIGNMENT\n=novalue\nKEY=value\n"
assert envfile.parse(text) == {"KEY": "value"}
def test_a_value_keeps_its_content_but_loses_surrounding_space() -> None:
assert envfile.parse(" A = one=two \n") == {"A": "one=two"}
def test_a_repeated_key_takes_the_last_one_as_dotenv_does() -> None:
assert envfile.parse("A=1\nA=2\n") == {"A": "2"}
def test_editing_a_present_key_replaces_it_where_it_stands() -> None:
rewritten = envfile.render(SAMPLE, {"COOLDOWN_SECONDS": "30"})
assert "COOLDOWN_SECONDS=30" in rewritten
# The whole reason this is not a regenerated file: the prose survives.
assert "# Empty means every group this account is in." in rewritten
assert rewritten.index("PHONE_NUMBER") < rewritten.index("GROUP_IDS")
assert "Added by the management page" not in rewritten
def test_a_key_the_file_never_had_is_appended_not_inserted() -> None:
rewritten = envfile.render(SAMPLE, {"LOG_LEVEL": "DEBUG"})
# At the end, under a marker: the comments in the body explain their own
# key, and dropping a new one among them would attach it to prose about
# something else entirely.
assert rewritten.endswith("# Added by the management page.\nLOG_LEVEL=DEBUG\n")
def test_the_file_is_returned_unchanged_when_nothing_was_edited() -> None:
assert envfile.render(SAMPLE, {}) == SAMPLE
def test_the_candidate_is_the_file_with_the_edits_on_top() -> None:
merged = envfile.candidate(SAMPLE, {"GROUP_IDS": "group.abc"})
assert merged["GROUP_IDS"] == "group.abc"
assert merged["PHONE_NUMBER"] == "+15550000000"
def test_the_process_environment_shows_through_where_the_file_is_silent(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# load_dotenv does not overwrite what systemd already exported, so a check
# that ignored os.environ would be validating a file the running unit is
# not actually configured by.
monkeypatch.setenv("SIGNAL_BOT_ONLY_IN_THE_ENVIRONMENT", "seen")
assert envfile.candidate(SAMPLE, {})["SIGNAL_BOT_ONLY_IN_THE_ENVIRONMENT"] == "seen"
def test_an_edit_wins_over_the_process_environment(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("GROUP_IDS", "group.from.systemd")
merged = envfile.candidate(SAMPLE, {"GROUP_IDS": "group.from.form"})
assert merged["GROUP_IDS"] == "group.from.form"
def test_writing_replaces_the_file_and_keeps_it_unreadable_to_others(
tmp_path: Path,
) -> None:
path = tmp_path / ".env"
path.write_text(SAMPLE, encoding="utf-8")
path.chmod(0o600)
envfile.write(path, {"COOLDOWN_SECONDS": "45"})
assert "COOLDOWN_SECONDS=45" in path.read_text(encoding="utf-8")
assert path.stat().st_mode & 0o777 == envfile.FILE_MODE
# Temp-then-replace, and nothing left behind: this file is edited from a
# browser tab that can be closed mid-save, and a half-written .env is not
# a bad value, it is a bot that will not start.
assert list(tmp_path.iterdir()) == [path]

View File

@ -0,0 +1,86 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Which .env keys a browser may write, and what makes an edit valid."""
from __future__ import annotations
import pytest
from signal_bot.groups import ConfigError
from signal_bot.manage import envkeys
SAMPLE = """\
PHONE_NUMBER=+15550000000
GROUP_IDS=
COOLDOWN_SECONDS=10
"""
def test_every_known_key_is_listed_with_whether_it_may_be_edited() -> None:
listed = envkeys.settings(SAMPLE)
by_key = {entry["key"]: entry for entry in listed}
assert by_key["COOLDOWN_SECONDS"] == {
"key": "COOLDOWN_SECONDS",
"value": "10",
"editable": True,
}
# Absent from the file is not absent from the page: empty is how .env
# spells "use the default", and the page has to be able to show that.
assert by_key["SILENCE_SECONDS"]["value"] == ""
assert len(listed) == len(envkeys.EDITABLE) + len(envkeys.READ_ONLY)
def test_the_account_and_the_container_it_talks_to_are_not_editable() -> None:
# PHONE_NUMBER because the container holds the matching registration.
# API_BASE_URL because repointing it sends every outgoing message and
# every attachment fetch somewhere else on the next restart, which is a
# strictly worse hole than changing the number.
assert "PHONE_NUMBER" in envkeys.READ_ONLY
assert "API_BASE_URL" in envkeys.READ_ONLY
assert not set(envkeys.EDITABLE) & set(envkeys.READ_ONLY)
def test_a_read_only_key_is_named_as_the_reason_for_refusing() -> None:
assert envkeys.refuse({"PHONE_NUMBER": "+15550000001"}, groups_known=True) == "PHONE_NUMBER"
def test_an_ordinary_edit_is_allowed() -> None:
assert envkeys.refuse({"COOLDOWN_SECONDS": "20"}, groups_known=True) is None
def test_group_keys_are_refused_while_the_container_is_unreachable() -> None:
# parse_group_ids only checks syntax. Startup additionally requires that
# every listed group is one the bot actually joined, judged against a
# listing that was really read -- so writing one blind turns a scoping
# mistake into a unit that will not start.
assert envkeys.refuse({"GROUP_IDS": "group.x"}, groups_known=False) == "GROUP_IDS"
assert envkeys.refuse({"SELFTEST_GROUPS": "x"}, groups_known=False) == "SELFTEST_GROUPS"
def test_group_keys_are_allowed_once_a_listing_has_been_read() -> None:
assert envkeys.refuse({"GROUP_IDS": "group.x"}, groups_known=True) is None
def test_a_valid_edit_is_one_the_bot_could_start_on() -> None:
envkeys.validate(SAMPLE, {"COOLDOWN_SECONDS": "20"})
def test_an_edit_the_bot_would_refuse_at_boot_is_refused_at_the_form() -> None:
# The message is load_config's own, so the page shows what the journal
# would have shown eight hours later.
with pytest.raises(ConfigError, match="REPORT_HOUR"):
envkeys.validate(SAMPLE, {"REPORT_HOUR": "99"})
def test_a_non_numeric_cooldown_is_refused() -> None:
with pytest.raises(ConfigError, match="COOLDOWN_SECONDS"):
envkeys.validate(SAMPLE, {"COOLDOWN_SECONDS": "soon"})
def test_an_unknown_log_level_is_refused() -> None:
with pytest.raises(ConfigError, match="LOG_LEVEL"):
envkeys.validate(SAMPLE, {"LOG_LEVEL": "LOUD"})
def test_a_trusted_inviter_that_is_neither_a_number_nor_a_uuid_is_refused() -> None:
with pytest.raises(ConfigError, match="TRUSTED_INVITERS"):
envkeys.validate(SAMPLE, {"TRUSTED_INVITERS": "whoever"})