Gabriel Brown
fd99569666
Listing zones and services is what firewall-cmd already does. The question it does not answer needs both halves at once: a port is reachable only when something is LISTENING on a network address AND the firewall permits it. On this machine that crossing is the whole story. The rules look unremarkable -- one zone, three services, a port range -- and what they mean is that PostgreSQL and Redis, published by rootless containers on every interface, are reachable by anyone on the network. Neither half says that alone, which is exactly how a tidy rules list coexists with an open database. Nothing was misconfigured: Fedora's default zone met podman's default publish behaviour. Ephemeral client sockets are excluded. A browser's outbound UDP port is indistinguishable from a service in ss, and listing twenty of them buried the two rows that mattered. Closing the port range names what it would cut off, by service, before doing it, and removing ssh says so when someone is connected over it. Rich rules are shown and never edited: a syntax is not a setting, but hiding it would misrepresent the configuration. The contract needed a recorded firewall, and the reason is worth keeping. The rule this page exists for cannot be tested against this machine -- its zone permits everything above 1024, so "listening" and "listening and permitted" give identical answers, and a blocked listener needs a port below 1024, which needs root. With the crossing deleted, the contract passed. It now runs against a fixture where two listeners are blocked, and catches it. Also here: polkit response files are written 0600 rather than at the default mask, the agent sweeps requests left by an instance that did not exit cleanly, and the write sweep waits for its harness to be ready instead of reporting the startup race as settings that failed. Claude-Session: https://claude.ai/code/session_01BRvzt4H8XXLPVH5MyYdk9L
308 lines
12 KiB
Python
308 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
|
|
"""Flip every compositor-backed setting, read it back, and put it back.
|
|
|
|
The existing write contract does this for five settings. There are sixty-five,
|
|
and the ones it does not cover are exactly where a silently-dead setting would
|
|
live: nothing fails, nothing logs, the row just does not do anything.
|
|
|
|
Each setting is driven through SystemSettings.commitPreference -- the same entry
|
|
point a settings row uses -- so this exercises the real routing, the real
|
|
compositor write, and the real read-back verification, not a shape.
|
|
|
|
Every original value is captured before anything changes and restored at the
|
|
end, through hyprctl directly, so a broken write path cannot also break the
|
|
restore.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
SCHEMA = "config/dot/quickshell/config/PreferenceSchema.qml"
|
|
|
|
# Read-back shapes this sweep knows how to compare. A gradient or a vec2 has no
|
|
# single scalar to diff, and guessing one would produce false failures; those
|
|
# are reported as skipped rather than quietly counted as verified.
|
|
COMPARABLE = {"bool", "int", "float", "str"}
|
|
|
|
# Settings whose value this must not choose freely.
|
|
#
|
|
# kb_layout is the one that can lock someone out of their own keyboard if a
|
|
# sweep picks a layout they cannot type in, and the existing write contract
|
|
# already covers it deliberately.
|
|
SKIP_KEYS = {"keyboardLayout", "keyboardVariant", "keyboardOptions", "keyboardModel"}
|
|
|
|
|
|
def run(command: list[str], timeout: float = 20.0) -> subprocess.CompletedProcess:
|
|
return subprocess.run(command, capture_output=True, text=True,
|
|
timeout=timeout, check=False)
|
|
|
|
|
|
def local_entries() -> list[dict]:
|
|
"""User-facing settings that Panama stores itself.
|
|
|
|
These never reach the compositor, so the compositor cannot be asked whether
|
|
they took. What can be asked is whether the value came back out of the
|
|
preference store -- a setting that silently fails to persist is the same
|
|
dead row from the user's side.
|
|
"""
|
|
source = open(SCHEMA, encoding="utf-8").read()
|
|
entries = []
|
|
for chunk in source.split("key: ")[1:]:
|
|
head = chunk[:1400]
|
|
if re.search(r"internal:\s*true", head) or re.search(r"hypr:\s*\{", head):
|
|
continue
|
|
key = re.match(r'"([A-Za-z0-9_]+)"', chunk)
|
|
kind = re.search(r'type:\s*"([a-z]+)"', head)
|
|
if not (key and kind) or kind.group(1) == "json":
|
|
continue
|
|
entry = {"key": key.group(1), "type": kind.group(1)}
|
|
for bound in ("min", "max", "step"):
|
|
found = re.search(rf"\b{bound}:\s*(-?[0-9.]+)", head)
|
|
if found:
|
|
entry[bound] = float(found.group(1))
|
|
options = re.search(r"options:\s*\[(.*?)\]", head, re.S)
|
|
if options:
|
|
entry["values"] = [
|
|
ast.literal_eval(value) if value.lstrip().startswith(("'", '"'))
|
|
else int(value)
|
|
for value in re.findall(r"value:\s*(\"[^\"]*\"|-?\d+)", options.group(1))
|
|
]
|
|
entries.append(entry)
|
|
return entries
|
|
|
|
|
|
def schema_entries() -> list[dict]:
|
|
"""Every user-facing setting that maps onto a compositor option."""
|
|
source = open(SCHEMA, encoding="utf-8").read()
|
|
entries = []
|
|
for chunk in source.split("key: ")[1:]:
|
|
head = chunk[:1400]
|
|
if re.search(r"internal:\s*true", head):
|
|
continue
|
|
hypr = re.search(r"hypr:\s*\{(.*?)\n\s*\}", head, re.S)
|
|
if not hypr:
|
|
continue
|
|
key = re.match(r'"([A-Za-z0-9_]+)"', chunk)
|
|
kind = re.search(r'type:\s*"([a-z]+)"', head)
|
|
option = re.search(r'option:\s*"([^"]+)"', hypr.group(1))
|
|
read_as = re.search(r'readAs:\s*"([a-z]+)"', hypr.group(1))
|
|
if not (key and kind and option):
|
|
continue
|
|
entry = {
|
|
"key": key.group(1),
|
|
"type": kind.group(1),
|
|
"option": option.group(1),
|
|
"readAs": read_as.group(1) if read_as else "int",
|
|
"invert": bool(re.search(r"invert:\s*true", hypr.group(1))),
|
|
}
|
|
for bound in ("min", "max", "step"):
|
|
found = re.search(rf"\b{bound}:\s*(-?[0-9.]+)", head)
|
|
if found:
|
|
entry[bound] = float(found.group(1))
|
|
options = re.search(r"options:\s*\[(.*?)\]", head, re.S)
|
|
if options:
|
|
entry["values"] = [
|
|
ast.literal_eval(value) if value.lstrip().startswith(("'", '"'))
|
|
else int(value)
|
|
for value in re.findall(r"value:\s*(\"[^\"]*\"|-?\d+)", options.group(1))
|
|
]
|
|
entries.append(entry)
|
|
return entries
|
|
|
|
|
|
def read_option(entry: dict):
|
|
"""What the compositor currently holds, in the shape the schema declares."""
|
|
result = run(["hyprctl", "-j", "getoption", entry["option"]])
|
|
if result.returncode != 0:
|
|
return None
|
|
try:
|
|
payload = json.loads(result.stdout)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
field = {"bool": "bool", "int": "int", "float": "float", "str": "str"}.get(entry["readAs"])
|
|
if field is None or field not in payload:
|
|
return None
|
|
return payload[field]
|
|
|
|
|
|
def stored_from_option(entry: dict, raw):
|
|
"""The value a preference would hold for this compositor reading."""
|
|
if entry["readAs"] == "bool":
|
|
value = bool(raw)
|
|
return (not value) if entry["invert"] else value
|
|
if entry["readAs"] == "int" and entry["type"] == "bool":
|
|
value = bool(raw)
|
|
return (not value) if entry["invert"] else value
|
|
return raw
|
|
|
|
|
|
def pick_target(entry: dict, current):
|
|
"""A valid value this setting does not currently hold."""
|
|
if entry["type"] == "bool":
|
|
return not bool(current)
|
|
if entry["type"] == "enum":
|
|
for candidate in entry.get("values", []):
|
|
if candidate != current:
|
|
return candidate
|
|
return None
|
|
if entry["type"] in ("int", "real"):
|
|
low = entry.get("min", 0.0)
|
|
high = entry.get("max", max(low + 1.0, float(current or 0) + 1.0))
|
|
step = entry.get("step", 1.0)
|
|
candidate = float(current or 0) + step
|
|
if candidate > high:
|
|
candidate = float(current or 0) - step
|
|
if candidate < low:
|
|
candidate = low if low != current else high
|
|
if entry["type"] == "int":
|
|
candidate = int(round(candidate))
|
|
if candidate == current:
|
|
return None
|
|
return candidate
|
|
return None
|
|
|
|
|
|
def ipc(config_home: str, harness: str, *arguments: str) -> subprocess.CompletedProcess:
|
|
environment = dict(os.environ, XDG_CONFIG_HOME=config_home)
|
|
return subprocess.run(["qs", "-p", harness, "ipc", "call", "settings-system-test", *arguments],
|
|
capture_output=True, text=True, timeout=30, check=False,
|
|
env=environment)
|
|
|
|
|
|
def main() -> int:
|
|
# Driven by settings-write-sweep-contract.sh, which starts the harness this
|
|
# needs and restores the desktop afterwards. Run bare -- by a suite runner
|
|
# walking the directory, say -- it says so instead of failing on argv.
|
|
if len(sys.argv) < 3:
|
|
print("settings_write_sweep is driven by tests/quickshell/"
|
|
"settings-write-sweep-contract.sh; run that instead.")
|
|
return 0
|
|
|
|
config_home = sys.argv[1]
|
|
harness = sys.argv[2]
|
|
|
|
entries = [entry for entry in schema_entries() if entry["key"] not in SKIP_KEYS]
|
|
verified, skipped, failures, originals = [], [], [], {}
|
|
|
|
for entry in entries:
|
|
raw = read_option(entry)
|
|
if raw is None or entry["readAs"] not in COMPARABLE:
|
|
skipped.append((entry["key"], f"read-back shape {entry['readAs']}"))
|
|
continue
|
|
|
|
current = stored_from_option(entry, raw)
|
|
originals[entry["key"]] = (entry, raw)
|
|
target = pick_target(entry, current)
|
|
if target is None:
|
|
skipped.append((entry["key"], "no second valid value"))
|
|
continue
|
|
|
|
# Whatever happens below, this setting goes back to where it was
|
|
# before the next one is touched. A sweep that dies holding sixty
|
|
# settings at test values would be worse than no sweep.
|
|
try:
|
|
answer = ipc(config_home, harness, "commit", entry["key"], json.dumps(target))
|
|
if answer.returncode != 0:
|
|
failures.append((entry["key"], f"commit failed: {answer.stderr.strip()[:80]}"))
|
|
continue
|
|
if answer.stdout.strip().lower() not in ("true", ""):
|
|
failures.append((entry["key"], f"commit refused the value ({answer.stdout.strip()[:40]})"))
|
|
continue
|
|
|
|
# The compositor is asked again for up to a second. commitPreference
|
|
# verifies before storing, so a correct write is usually visible
|
|
# immediately -- but "usually" makes a flaky test, and a flaky test
|
|
# is worse than none. A write that never happens never matches, no
|
|
# matter how long this waits.
|
|
after = None
|
|
matched = False
|
|
deadline = time.monotonic() + 1.0
|
|
while True:
|
|
after = stored_from_option(entry, read_option(entry))
|
|
if isinstance(target, float) or isinstance(after, float):
|
|
matched = after is not None and abs(float(after) - float(target)) < 0.02
|
|
else:
|
|
matched = after == target
|
|
if matched or time.monotonic() > deadline:
|
|
break
|
|
time.sleep(0.05)
|
|
if not matched:
|
|
failures.append((entry["key"],
|
|
f"set to {target!r} but the compositor reports {after!r} "
|
|
f"({entry['option']})"))
|
|
continue
|
|
verified.append(entry["key"])
|
|
finally:
|
|
ipc(config_home, harness, "commit", entry["key"], json.dumps(current))
|
|
|
|
# ── Settings Panama stores itself ────────────────────────────────────────
|
|
# The harness answers its IPC socket slightly before it can serve queries,
|
|
# and a read in that window comes back "Not ready to accept queries yet" --
|
|
# which is a race in this test, not a setting that failed to round-trip.
|
|
for _ in range(40):
|
|
probe = ipc(config_home, harness, "stored", "use24Hour")
|
|
if probe.returncode == 0 and "not ready" not in probe.stdout.lower():
|
|
break
|
|
time.sleep(0.25)
|
|
|
|
local_verified, local_skipped, local_failures = [], [], []
|
|
for entry in local_entries():
|
|
raw = ipc(config_home, harness, "stored", entry["key"])
|
|
if raw.returncode != 0:
|
|
local_failures.append((entry["key"], "could not be read from the store"))
|
|
continue
|
|
try:
|
|
current = json.loads(raw.stdout.strip() or "null")
|
|
except json.JSONDecodeError:
|
|
local_failures.append((entry["key"], f"stored value is not readable: {raw.stdout.strip()[:40]}"))
|
|
continue
|
|
|
|
target = pick_target(entry, current)
|
|
if target is None:
|
|
local_skipped.append((entry["key"], "no second valid value"))
|
|
continue
|
|
|
|
try:
|
|
answer = ipc(config_home, harness, "commit", entry["key"], json.dumps(target))
|
|
if answer.returncode != 0 or answer.stdout.strip().lower() not in ("true", ""):
|
|
local_failures.append((entry["key"], "the store refused the value"))
|
|
continue
|
|
back = ipc(config_home, harness, "stored", entry["key"])
|
|
try:
|
|
after = json.loads(back.stdout.strip() or "null")
|
|
except json.JSONDecodeError:
|
|
after = None
|
|
if isinstance(target, float) or isinstance(after, float):
|
|
matched = after is not None and abs(float(after) - float(target)) < 0.02
|
|
else:
|
|
matched = after == target
|
|
if not matched:
|
|
local_failures.append((entry["key"], f"set to {target!r} but the store reports {after!r}"))
|
|
continue
|
|
local_verified.append(entry["key"])
|
|
finally:
|
|
ipc(config_home, harness, "commit", entry["key"], json.dumps(current))
|
|
|
|
print(json.dumps({
|
|
"verified": verified,
|
|
"skipped": skipped,
|
|
"failures": failures,
|
|
"total": len(entries),
|
|
"localVerified": local_verified,
|
|
"localSkipped": local_skipped,
|
|
"localFailures": local_failures,
|
|
}))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|