From 30cb46ca1fb157ce8b28e155b92bdaacf0ebda87 Mon Sep 17 00:00:00 2001 From: twilightsaurus Date: Tue, 6 Oct 2026 10:22:33 +0300 Subject: [PATCH] fix: support private-group Direct WSS parents --- deploy/README.md | 27 +- scripts/test-server-bundle.sh | 63 +++++ services/relay/src/imp_relay/direct_wss.py | 102 +++++-- services/relay/tests/test_direct_wss.py | 300 +++++++++++++++++++++ 4 files changed, 472 insertions(+), 20 deletions(-) diff --git a/deploy/README.md b/deploy/README.md index aae23ab..18491a2 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -215,16 +215,34 @@ Check without changing files, locks, credentials, or service state: ~/.local/bin/imp-direct-wss status ``` -Status reports configuration syntax validity, private-file safety, systemd -enabled/active state, and minimal health without disclosing credential -material. The fixed paths are `~/.config/imp/gateway.env`, +Status separately reports configuration path safety, syntax validity, file and +directory modes, gateway unit policy, systemd enabled/active state, and minimal +health without disclosing credential material. An unsafe path names the +offending directory and reason. The fixed paths are `~/.config/imp/gateway.env`, `~/.config/systemd/user/imp-gateway.service`, and `~/.local/share/imp/current/.venv/bin/imp-gateway`. The config's only accepted content is exactly one `IMP_GATEWAY_TOKEN_SHA256=<64 hex characters>` assignment, with full-line comments and surrounding whitespace allowed; NUL bytes are rejected anywhere, including comments. It is not a general environment file. The config directory and file must be owned by the target user with modes -`0700` and `0600`, respectively. +exactly `0700` and `0600`, respectively. + +The ancestors `~/.config`, `~/.config/systemd`, and `~/.config/systemd/user` +must be owned by the target user and never world-writable. They may be +group-writable only when their group is the target user's primary group and the +system account database enumeration (`getent passwd`/`getent group`) lists the +target user, no other account with that primary group, and no other +supplementary member in any group record with that GID. A group-writable +ancestor that also carries an extended access ACL is rejected. That covers the +ordinary Ubuntu local user-private-group layout (`0775` with umask `002`), so +those directories need no permission change. A group-writable ancestor whose +group is shared, or cannot be resolved or enumerated, is rejected. + +This check sees only enumerable accounts and groups. On hosts whose account +sources do not fully enumerate users and groups, such as some SSSD or LDAP +configurations, it is not proof that no other principal shares the group. If any +other account can write such an ancestor through directory-service membership +or another mechanism, remove group write from that ancestor. The normal release installer does not create pairing credentials or enable `imp-gateway.service`. Upgrade preserves the gateway's existing enablement and @@ -658,6 +676,7 @@ curl -s http://127.0.0.1:8788/healthz | `imp-feed` exits immediately with "another Imp feed holds ..." | A duplicate instance is running - manual invocation while the service is active, or a second service instance. `systemctl --user status imp-feed.service`, then stop the extra process. | | Relay reachable but no HUD data | In SSH mode, check `curl http://127.0.0.1:8787/healthz` through the forward. In Direct mode, check both local gateway health and the public TLS `/healthz`; then confirm a relay producer is attached and inspect `imp-feed.service`. | | Direct WSS repeatedly reconnects | Confirm the public certificate is trusted/current, the reverse proxy forwards `/state` and `/action` to `127.0.0.1:8788`, and the desktop pairing token matches the digest in `gateway.env`. Never move the token into the URL while debugging. | +| Direct WSS path is unsafe or unit policy unsupported | `imp-direct-wss status` names the directory and reason. For a shared group-writable or world-writable ancestor, remove that write access or give the directory the user's private group; do not loosen `~/.config/imp` or `gateway.env`. | | TinyFugue shows an `fwrite` error line | The feed is down or the hook symlink target directory is missing. TinyFugue is not blocked by this - it is the intended fail-open behaviour - but no HUD update reaches the relay until the feed is running again. | | Services do not survive a reboot | Confirm `loginctl show-user "$USER" -p Linger` reports `Linger=yes`; without it, user units never start without an interactive login. | | Relay bound to more than loopback | The current relay refuses every non-loopback host and has no override. Restore the shipped unit and executable if this occurs. | diff --git a/scripts/test-server-bundle.sh b/scripts/test-server-bundle.sh index b592a70..6845f47 100755 --- a/scripts/test-server-bundle.sh +++ b/scripts/test-server-bundle.sh @@ -761,6 +761,69 @@ try: assert inactive_token not in status.stdout and inactive_token not in status.stderr assert config_path.stat().st_mtime_ns == status_before.st_mtime_ns assert (config_path.parent / "gateway.env.lock").stat().st_mtime_ns == lock_before.st_mtime_ns + + # Ubuntu user-private-group layout: 0775 ancestors under a 0700 ~/.config/imp. + # The account database is substituted inside the installed interpreter so the + # verdict cannot depend on the runner's real passwd/group contents. + ancestors = {path: path.stat().st_mode & 0o7777 for path in (home / ".config", home / ".config/systemd", home / ".config/systemd/user")} + account_cli = ( + "import grp, os, pwd, sys\n" + "from imp_relay import direct_wss\n" + "uid, gid = os.geteuid(), os.stat(os.path.expanduser('~/.config')).st_gid\n" + "user = pwd.struct_passwd(('imp-user', 'x', uid, gid, '', '', ''))\n" + "group = grp.struct_group(('imp-user', 'x', gid, sys.argv[2:]))\n" + "pwd.getpwuid = lambda value: {uid: user}[value]\n" + "pwd.getpwall = lambda: [user]\n" + "grp.getgrgid = lambda value: {gid: group}[value]\n" + "grp.getgrall = lambda: [group]\n" + "raise SystemExit(direct_wss.main([sys.argv[1]]))\n" + ) + shared = "~/.config is group-writable by a group not shown private to the target user" + try: + for directory in ancestors: + directory.chmod(0o775) + for members, path_line, syntax_line, policy_line in ( + ( + ["imp-user"], + "gateway.env path: supported", + "gateway.env syntax: valid", + "gateway unit policy: supported", + ), + ( + ["imp-user", "other"], + f"gateway.env path: unsafe ({shared})", + "gateway.env syntax: not inspected", + f"gateway unit policy: unsupported ({shared})", + ), + ): + private_status = subprocess.run( + [str(runtime / "python"), "-c", account_cli, "status", *members], + env=environment, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + assert private_status.returncode == 0, private_status.stderr + lines = private_status.stdout.splitlines() + assert path_line in lines and syntax_line in lines and policy_line in lines, lines + assert inactive_token.decode() not in private_status.stdout + private_status.stderr + assert config_path.stat().st_mtime_ns == status_before.st_mtime_ns + + # Installed-code rotate under the same layout (gateway enabled + stopped here). + before_rotate = config_path.read_bytes() + private_token = token_from(run_tty([runtime / "python", "-c", account_cli, "rotate", "imp-user"])) + assert private_token != inactive_token + assert config_path.read_bytes() != before_rotate + assert private_token not in config_path.read_bytes() + assert config_path.stat().st_mode & 0o7777 == 0o600 + assert config_path.parent.stat().st_mode & 0o7777 == 0o700 + assert not active("imp-gateway.service") and state["enabled"] + finally: + for directory, mode in ancestors.items(): + directory.chmod(mode) + print("private-group ancestor status and rotate: OK") finally: with state_lock: stop_gateway() diff --git a/services/relay/src/imp_relay/direct_wss.py b/services/relay/src/imp_relay/direct_wss.py index dd97dbb..6d7000d 100644 --- a/services/relay/src/imp_relay/direct_wss.py +++ b/services/relay/src/imp_relay/direct_wss.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import errno import http.client import json import os @@ -27,6 +28,8 @@ try: import fcntl + import grp + import pwd except ImportError: # Keep desktop imports/builds working outside Linux. fcntl = None # type: ignore[assignment] @@ -97,6 +100,7 @@ class ConfigInspection: digest: bytes | None data: bytes | None exists: bool + path: str = "supported" def _supported_platform() -> None: @@ -139,7 +143,47 @@ def _open_home_component(parent_fd: int, name: str, uid: int) -> int: return fd -def _open_directory(parent_fd: int, name: str, uid: int, *, create: bool = False) -> int: +def _private_primary_group(gid: int, uid: int) -> bool: + """Whether the enumerable account database shows gid as the target user's unshared primary group. + + Only accounts and groups visible through pwd/grp enumeration are considered; account sources + that do not enumerate fully can still grant access this check cannot see. + """ + try: + user = pwd.getpwuid(uid) + grp.getgrgid(gid) + groups = [group for group in grp.getgrall() if group.gr_gid == gid] + accounts = pwd.getpwall() + except (KeyError, OSError): + return False + # Requiring the target user in the enumeration rejects account sources getpwall cannot list. + return ( + user.pw_gid == gid + and bool(groups) + and any(account.pw_uid == uid and account.pw_gid == gid for account in accounts) + and all(account.pw_uid == uid for account in accounts if account.pw_gid == gid) + and all(member == user.pw_name for group in groups for member in group.gr_mem) + ) + + +def _access_acl(fd: int) -> bool: + """Whether the directory carries an extended access ACL; inspection errors fail closed.""" + try: + os.getxattr(fd, "system.posix_acl_access") + except OSError as error: + return error.errno not in {errno.ENODATA, errno.ENOTSUP, errno.EOPNOTSUPP} + return True + + +def _open_directory( + parent_fd: int, + name: str, + uid: int, + *, + display: str, + create: bool = False, + allow_private_group: bool = False, +) -> int: created = False try: fd = os.open(name, _directory_flags(), dir_fd=parent_fd) @@ -154,18 +198,32 @@ def _open_directory(parent_fd: int, name: str, uid: int, *, create: bool = False try: fd = os.open(name, _directory_flags(), dir_fd=parent_fd) except OSError: - raise DirectWssError("configuration path contains an unsafe directory") from None + raise DirectWssError(f"{display} is not a safe directory") from None except OSError: - raise DirectWssError("configuration path contains an unsafe directory") from None + raise DirectWssError(f"{display} is not a safe directory") from None try: info = os.fstat(fd) except OSError: os.close(fd) - raise DirectWssError("configuration path contains an unsafe directory") from None + raise DirectWssError(f"{display} is not a safe directory") from None mode = stat.S_IMODE(info.st_mode) - if not stat.S_ISDIR(info.st_mode) or info.st_uid != uid or mode & 0o022: + problem = None + if not stat.S_ISDIR(info.st_mode): + problem = "is not a safe directory" + elif info.st_uid != uid: + problem = "is not owned by the target user" + elif mode & 0o002: + problem = "is world-writable" + elif mode & 0o020 and not allow_private_group: + # Only ~/.config/imp opens without the exception; its policy is exactly 0700. + problem = "must be owned by the target user with mode 0700" + elif mode & 0o020 and not _private_primary_group(info.st_gid, uid): + problem = "is group-writable by a group not shown private to the target user" + elif mode & 0o020 and _access_acl(fd): + problem = "is group-writable and has an extended access ACL that could not be excluded" + if problem is not None: os.close(fd) - raise DirectWssError("configuration path contains an unsafe directory") + raise DirectWssError(f"{display} {problem}") if created: try: os.fchmod(fd, 0o700) @@ -194,9 +252,11 @@ def _open_home() -> Iterator[tuple[int, str, int]]: def _config_directory(home_fd: int, uid: int, *, create: bool, private: bool = True) -> int: - config_fd = _open_directory(home_fd, ".config", uid, create=create) + config_fd = _open_directory( + home_fd, ".config", uid, display="~/.config", create=create, allow_private_group=True + ) try: - imp_fd = _open_directory(config_fd, "imp", uid, create=create) + imp_fd = _open_directory(config_fd, "imp", uid, display="~/.config/imp", create=create) finally: os.close(config_fd) if private and stat.S_IMODE(os.fstat(imp_fd).st_mode) != 0o700: @@ -385,13 +445,17 @@ def _property_map(output: str) -> dict[str, str]: def _unit_file(home_fd: int, uid: int, home: str) -> str: try: - config_fd = _open_directory(home_fd, ".config", uid) + config_fd = _open_directory(home_fd, ".config", uid, display="~/.config", allow_private_group=True) try: - systemd_fd = _open_directory(config_fd, "systemd", uid) + systemd_fd = _open_directory( + config_fd, "systemd", uid, display="~/.config/systemd", allow_private_group=True + ) finally: os.close(config_fd) try: - user_fd = _open_directory(systemd_fd, "user", uid) + user_fd = _open_directory( + systemd_fd, "user", uid, display="~/.config/systemd/user", allow_private_group=True + ) finally: os.close(systemd_fd) except FileNotFoundError: @@ -714,8 +778,10 @@ def _status_config(home_fd: int, uid: int) -> ConfigInspection: config_fd = _config_directory(home_fd, uid, create=False, private=False) except FileNotFoundError: return ConfigInspection("missing", "unavailable", "unavailable", None, None, False) - except DirectWssError: - return ConfigInspection("unsafe", "unavailable", "unavailable", None, None, False) + except DirectWssError as error: + return ConfigInspection( + "not inspected", "unavailable", "unavailable", None, None, False, f"unsafe ({error})" + ) try: directory_value = stat.S_IMODE(os.fstat(config_fd).st_mode) directory_mode = f"{directory_value:04o} {'private' if directory_value == 0o700 else 'not private'}" @@ -723,14 +789,16 @@ def _status_config(home_fd: int, uid: int) -> ConfigInspection: data, info = _read_regular_file(config_fd, _CONFIG_NAME, uid, max_bytes=_MAX_CONFIG_BYTES) except FileNotFoundError: return ConfigInspection("missing", "unavailable", directory_mode, None, None, False) - except DirectWssError: + except DirectWssError as error: try: info = os.stat(_CONFIG_NAME, dir_fd=config_fd, follow_symlinks=False) mode = stat.S_IMODE(info.st_mode) mode_text = f"{mode:04o} {'private' if mode == 0o600 else 'not private'}" except OSError: mode_text = "unavailable" - return ConfigInspection("unsafe", mode_text, directory_mode, None, None, True) + return ConfigInspection( + "not inspected", mode_text, directory_mode, None, None, True, f"unsafe ({error})" + ) mode = stat.S_IMODE(info.st_mode) mode_text = f"{mode:04o} {'private' if mode == 0o600 else 'not private'}" try: @@ -751,7 +819,8 @@ def _status_service(home_fd: int, home: str) -> tuple[str, str, str]: enabled = "yes" if state.enabled else "no" active = "yes" if state.active else "no" policy = "supported" - except DirectWssError: + except DirectWssError as error: + policy = f"unsupported ({error})" try: shown = _systemctl("show", _UNIT, "--property=" + ",".join(_PROPERTIES)) except DirectWssError: @@ -773,6 +842,7 @@ def _status(stdout: IO[str]) -> None: inspection = _status_config(home_fd, uid) enabled, active, policy = _status_service(home_fd, home) health = "healthy" if active == "yes" and _gateway_health() else "unavailable" + stdout.write(f"gateway.env path: {inspection.path}\n") stdout.write(f"gateway.env syntax: {inspection.syntax}\n") stdout.write(f"gateway.env mode: {inspection.mode}\n") stdout.write(f"gateway.env directory mode: {inspection.directory_mode}\n") diff --git a/services/relay/tests/test_direct_wss.py b/services/relay/tests/test_direct_wss.py index b1b2695..dee3c04 100644 --- a/services/relay/tests/test_direct_wss.py +++ b/services/relay/tests/test_direct_wss.py @@ -1,12 +1,16 @@ from __future__ import annotations import base64 +import errno +import grp import hashlib import http.server import io import os +import pwd import secrets import shlex +import shutil import stat import subprocess import sys @@ -548,11 +552,307 @@ def test_status_reports_malformed_config_without_exposing_content( monkeypatch.setattr(sys, "stdout", out) assert direct_wss.main(["status"]) == 0 + assert "gateway.env path: supported" in out.getvalue() assert "gateway.env syntax: invalid" in out.getvalue() assert "SECRETISH" not in out.getvalue() assert not (config.parent / direct_wss._LOCK_NAME).exists() +SHARED_GROUP = "is group-writable by a group not shown private to the target user" +ACL = "is group-writable and has an extended access ACL that could not be excluded" + + +def _accounts( + monkeypatch: pytest.MonkeyPatch, + gid: int, + *, + primary_gid: int | None = None, + others: tuple[tuple[str, int, int], ...] = (), + members: tuple[str, ...] = ("imp-user",), + listed: bool = True, + resolvable: bool = True, + aliases: tuple[tuple[str, tuple[str, ...]], ...] = (), + groups_enumerable: bool = True, +) -> None: + """Replace the account database deterministically; the runner's passwd/group never matter.""" + uid = os.geteuid() + user = pwd.struct_passwd( + ("imp-user", "x", uid, gid if primary_gid is None else primary_gid, "", "/", "/bin/sh") + ) + accounts = [user] if listed else [] + accounts += [ + pwd.struct_passwd((name, "x", other_uid, other_gid, "", "/", "/bin/sh")) + for name, other_uid, other_gid in others + ] + + def getpwuid(value: int) -> pwd.struct_passwd: + if not resolvable or value != uid: + raise KeyError(value) + return user + + def getgrgid(value: int) -> grp.struct_group: + if not resolvable or value != gid: + raise KeyError(value) + return grp.struct_group(("imp-user", "x", gid, list(members))) + + def getgrall() -> list[grp.struct_group]: + if not groups_enumerable: + raise OSError("group enumeration unavailable") + records = [("imp-user", members), *aliases] + return [grp.struct_group((name, "x", gid, list(names))) for name, names in records] + + monkeypatch.setattr(direct_wss, "pwd", SimpleNamespace(getpwuid=getpwuid, getpwall=lambda: accounts)) + monkeypatch.setattr(direct_wss, "grp", SimpleNamespace(getgrgid=getgrgid, getgrall=getgrall)) + + +@pytest.mark.parametrize( + ("mode", "allow", "accounts", "problem"), + [ + (0o755, True, {"resolvable": False}, None), + (0o775, True, {}, None), + (0o777, True, {}, "is world-writable"), + (0o775, True, {"primary_gid": -1}, SHARED_GROUP), + (0o775, True, {"others": (("other", 4242, None),)}, SHARED_GROUP), + (0o775, True, {"members": ("imp-user", "other")}, SHARED_GROUP), + (0o775, True, {"resolvable": False}, SHARED_GROUP), + (0o775, True, {"listed": False}, SHARED_GROUP), + (0o775, True, {"aliases": (("shared-alias", ("other",)),)}, SHARED_GROUP), + (0o775, True, {"groups_enumerable": False}, SHARED_GROUP), + (0o775, False, {}, "must be owned by the target user with mode 0700"), + ], + ids=[ + "0755", + "0775-private-primary-group", + "world-writable", + "non-primary-group", + "other-primary-account", + "other-supplementary-member", + "lookup-failure", + "target-not-enumerated", + "duplicate-gid-record-with-other-member", + "group-enumeration-failure", + "strict-directory", + ], +) +def test_ancestor_group_write_requires_proven_private_primary_group( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mode: int, + allow: bool, + accounts: dict[str, Any], + problem: str | None, +) -> None: + directory = tmp_path / "ancestor" + directory.mkdir() + os.chmod(directory, mode) + gid = directory.stat().st_gid + if "others" in accounts: + accounts["others"] = tuple((name, other_uid, gid) for name, other_uid, _ in accounts["others"]) + if accounts.get("primary_gid") == -1: + accounts["primary_gid"] = gid + 1 + _accounts(monkeypatch, gid, **accounts) + parent = os.open(tmp_path, os.O_RDONLY | os.O_DIRECTORY) + try: + if problem is None: + os.close( + direct_wss._open_directory( + parent, "ancestor", os.geteuid(), display="~/a", allow_private_group=allow + ) + ) + else: + with pytest.raises(direct_wss.DirectWssError, match=f"^~/a {problem}$"): + direct_wss._open_directory( + parent, "ancestor", os.geteuid(), display="~/a", allow_private_group=allow + ) + finally: + os.close(parent) + + +def test_ancestor_owned_by_another_user_is_rejected(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + (tmp_path / "ancestor").mkdir(mode=0o755) + _accounts(monkeypatch, (tmp_path / "ancestor").stat().st_gid) + parent = os.open(tmp_path, os.O_RDONLY | os.O_DIRECTORY) + try: + with pytest.raises(direct_wss.DirectWssError, match="^~/a is not owned by the target user$"): + direct_wss._open_directory( + parent, "ancestor", os.geteuid() + 1, display="~/a", allow_private_group=True + ) + finally: + os.close(parent) + + +@pytest.mark.parametrize( + ("error", "rejected"), + [ + (None, True), + (errno.ENODATA, False), + (errno.ENOTSUP, False), + (errno.EOPNOTSUPP, False), + (errno.EACCES, True), + (errno.EIO, True), + ], + ids=["acl-present", "no-acl", "enotsup", "eopnotsupp", "eacces", "eio"], +) +def test_private_group_exception_rejects_present_or_uninspectable_access_acl( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, error: int | None, rejected: bool +) -> None: + directory = tmp_path / "ancestor" + directory.mkdir() + os.chmod(directory, 0o775) + _accounts(monkeypatch, directory.stat().st_gid) + inspected: list[Any] = [] + + def getxattr(target: Any, name: str) -> bytes: + inspected.append((type(target), name)) + if error is not None: + raise OSError(error, os.strerror(error)) + return b"acl" + + monkeypatch.setattr(os, "getxattr", getxattr) + parent = os.open(tmp_path, os.O_RDONLY | os.O_DIRECTORY) + try: + if rejected: + with pytest.raises(direct_wss.DirectWssError, match=f"^~/a {ACL}$"): + direct_wss._open_directory( + parent, "ancestor", os.geteuid(), display="~/a", allow_private_group=True + ) + else: + os.close( + direct_wss._open_directory( + parent, "ancestor", os.geteuid(), display="~/a", allow_private_group=True + ) + ) + finally: + os.close(parent) + assert inspected == [(int, "system.posix_acl_access")] + + +def test_named_user_write_acl_rejects_private_group_ancestor( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + if shutil.which("setfacl") is None: + pytest.skip("setfacl unavailable") + directory = tmp_path / "ancestor" + directory.mkdir() + os.chmod(directory, 0o775) + _accounts(monkeypatch, directory.stat().st_gid) + parent = os.open(tmp_path, os.O_RDONLY | os.O_DIRECTORY) + try: + os.close( + direct_wss._open_directory( + parent, "ancestor", os.geteuid(), display="~/a", allow_private_group=True + ) + ) + if subprocess.run(["setfacl", "-m", "u:nobody:rwx", str(directory)], capture_output=True).returncode: + pytest.skip("filesystem does not support POSIX ACLs") + assert stat.S_IMODE(directory.stat().st_mode) == 0o775 + with pytest.raises(direct_wss.DirectWssError, match=f"^~/a {ACL}$"): + direct_wss._open_directory( + parent, "ancestor", os.geteuid(), display="~/a", allow_private_group=True + ) + finally: + os.close(parent) + + +def _ubuntu_layout(home: Path) -> int: + for relative in (".config", ".config/systemd", ".config/systemd/user"): + os.chmod(home / relative, 0o775) + return (home / ".config").stat().st_gid + + +def test_status_and_rotate_accept_ubuntu_private_group_ancestors( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + home = _home(tmp_path, monkeypatch) + config = _write_config(home, f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode()) + service = _fake_systemctl(home, monkeypatch, active=False, enabled=True) + _accounts(monkeypatch, _ubuntu_layout(home)) + out = io.StringIO() + monkeypatch.setattr(sys, "stdout", out) + + assert direct_wss.main(["status"]) == 0 + for line in ( + "gateway.env path: supported", + "gateway.env syntax: valid", + "gateway.env mode: 0600 private", + "gateway.env directory mode: 0700 private", + "gateway unit policy: supported", + ): + assert line + "\n" in out.getvalue() + assert token_digest(OLD_TOKEN).hex() not in out.getvalue() + + monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN) + stdout, _ = _tty_streams(monkeypatch) + assert direct_wss.main(["rotate"]) == 0 + assert direct_wss._parse_config(config.read_bytes()) == token_digest(TOKEN) + assert stat.S_IMODE(config.stat().st_mode) == 0o600 + assert stat.S_IMODE(config.parent.stat().st_mode) == 0o700 + assert not service.actions + + +@pytest.mark.parametrize("operation", ["status", "setup", "rotate"]) +def test_shared_group_config_ancestor_is_a_path_failure_not_syntax( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, operation: str +) -> None: + home = _home(tmp_path, monkeypatch) + content = f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode() + config = _write_config(home, content) + service = _fake_systemctl(home, monkeypatch, active=True, enabled=True) + _accounts(monkeypatch, _ubuntu_layout(home), members=("imp-user", "other")) + monkeypatch.setattr(secrets, "token_urlsafe", lambda length: pytest.fail("generated token")) + stdout, stderr = _tty_streams(monkeypatch) + + assert direct_wss.main([operation]) == (0 if operation == "status" else 1) + output = stdout.getvalue() + stderr.getvalue() + reason = f"~/.config {SHARED_GROUP}" + if operation == "status": + assert f"gateway.env path: unsafe ({reason})\n" in output + assert "gateway.env syntax: not inspected\n" in output + assert f"gateway unit policy: unsupported ({reason})\n" in output + else: + assert output == f"error: {reason}\n" + assert token_digest(OLD_TOKEN).hex() not in output and OLD_TOKEN not in output + assert config.read_bytes() == content + assert not (config.parent / direct_wss._LOCK_NAME).exists() + assert not service.actions + + +def test_unsafe_systemd_ancestor_is_reported_as_unit_policy( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + home = _home(tmp_path, monkeypatch) + _write_config(home, f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode()) + _fake_systemctl(home, monkeypatch) + _accounts(monkeypatch, _ubuntu_layout(home)) + os.chmod(home / ".config/systemd/user", 0o777) + out = io.StringIO() + monkeypatch.setattr(sys, "stdout", out) + + assert direct_wss.main(["status"]) == 0 + assert "gateway.env path: supported\n" in out.getvalue() + assert "gateway.env syntax: valid\n" in out.getvalue() + assert "gateway unit policy: unsupported (~/.config/systemd/user is world-writable)\n" in out.getvalue() + assert token_digest(OLD_TOKEN).hex() not in out.getvalue() + + +def test_private_group_exception_never_applies_to_imp_config_directory( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + home = _home(tmp_path, monkeypatch) + config = _write_config(home, f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode()) + _fake_systemctl(home, monkeypatch) + _accounts(monkeypatch, _ubuntu_layout(home)) + os.chmod(config.parent, 0o770) + stdout, stderr = _tty_streams(monkeypatch) + + assert direct_wss.main(["status"]) == 0 + assert "gateway.env path: unsafe (~/.config/imp must be owned by the target user with mode 0700)\n" in ( + stdout.getvalue() + ) + assert direct_wss.main(["setup"]) == 1 + assert stderr.getvalue() == "error: ~/.config/imp must be owned by the target user with mode 0700\n" + + def test_systemd_preflight_accepts_only_shipped_manager_and_disk_policy( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: