adminandClaude Opus 5.5
6862dea52a
- test_price_secret.py: the fixed-tariff note on the real /config/validate route, EN and NL, and none for a spaces-only profile (web.py lang wiring). - test_price_service.py: the start line is INFO, not WARNING; a dismiss that raises does not stop the cycle. - test_maint13/test_setup_only/test_watchdog: ignore only the dismiss of the old card, so a create with that id is still counted. - DECISIONS rule 7 quotes the settings note as it is; messages.py comment. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
413 lines
18 KiB
Python
413 lines
18 KiB
Python
"""PRICE-01 round 2 (F1): a `password` option never reaches a page.
|
|
`python3 test_price_secret.py`
|
|
|
|
The real web app (`web.build_app`) runs over a fake Supervisor that holds a
|
|
site on `price_service` with a token. The token string must be absent from
|
|
`/config`, `/config/model` (`options`, `running`, `ui`), `/config/validate`,
|
|
a good save's answer and every refused save's answer. A save that sends the
|
|
placeholder, an empty field or no key keeps the stored token; a new value
|
|
replaces it. The same holds for any option the schema calls a password, not
|
|
only this one name.
|
|
|
|
⚠️ MUTATING A MODULE THIS FILE GRADES? Purge `__pycache__` first, or use
|
|
`tools/mutation_run.sh`. See MUTATION.md.
|
|
"""
|
|
|
|
import asyncio
|
|
import copy
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.join(HERE, "tools"))
|
|
logging.disable(logging.WARNING)
|
|
|
|
from aiohttp import ClientSession, web as aioweb # noqa: E402
|
|
|
|
import translations_lint as TL # noqa: E402
|
|
import ux13_demo as D # noqa: E402
|
|
from app import configui, price, web # noqa: E402
|
|
from app.configui import SECRET_SHOWN # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
TOKEN = "tk-SECRET-0123456789-abcdef"
|
|
NEW_TOKEN = "tk-REPLACED-9876543210-zyxw"
|
|
SCHEMA = D.schema_ui(D.CONFIG)
|
|
TRANSLATIONS = {"en": TL.parse_translation_file(
|
|
os.path.join(HERE, "translations", "en.yaml"))}
|
|
SITE = dict(D.defaults(D.CONFIG), setpoint_entity="number.goodwe_setpoint",
|
|
soc_entity="sensor.soc", batt_entity="sensor.batt",
|
|
price_provider="price_service", price_service_token=TOKEN)
|
|
|
|
|
|
class Sup:
|
|
def __init__(self, options, schema=SCHEMA):
|
|
self.options, self.schema, self.writes = copy.deepcopy(options), schema, []
|
|
|
|
async def info(self):
|
|
return {"options": copy.deepcopy(self.options), "schema": self.schema,
|
|
"translations": TRANSLATIONS}
|
|
|
|
async def write_options(self, o):
|
|
self.writes.append(copy.deepcopy(o))
|
|
self.options = copy.deepcopy(o)
|
|
|
|
async def restart(self, c):
|
|
raise AssertionError
|
|
|
|
|
|
class Hass:
|
|
session = None
|
|
token = ""
|
|
|
|
async def limits(self, _e):
|
|
return (-5000.0, 5000.0, 1.0)
|
|
|
|
async def states(self):
|
|
return []
|
|
|
|
|
|
class Ctl:
|
|
def __init__(self, o):
|
|
self.o, self.hass, self.inverters = dict(o), Hass(), []
|
|
|
|
def log_event(self, *a, **k):
|
|
pass
|
|
|
|
def status_payload(self):
|
|
return {"banner": "", "level": "ok", "rows": [], "checks": []}
|
|
|
|
|
|
async def serve(sup, ctl_options=None):
|
|
runner = aioweb.AppRunner(web.build_app(Ctl(ctl_options or sup.options), sup))
|
|
await runner.setup()
|
|
site = aioweb.TCPSite(runner, "127.0.0.1", 0)
|
|
await site.start()
|
|
port = site._server.sockets[0].getsockname()[1]
|
|
return runner, f"http://127.0.0.1:{port}/"
|
|
|
|
|
|
async def scenario():
|
|
out = {}
|
|
sup = Sup(SITE)
|
|
runner, root = await serve(sup)
|
|
seen = [] # every answer body, as text
|
|
try:
|
|
async with ClientSession() as s:
|
|
async def get(path):
|
|
async with s.get(root + path) as r:
|
|
t = await r.text()
|
|
seen.append(t)
|
|
return r.status, t
|
|
|
|
async def post(path, body):
|
|
async with s.post(root + path, json=body) as r:
|
|
t = await r.text()
|
|
seen.append(t)
|
|
return r.status, t
|
|
|
|
_, page = await get("config")
|
|
out["page"] = page
|
|
_, raw = await get("config/model?lang=en")
|
|
m = json.loads(raw)
|
|
out["model"] = m
|
|
out["raw_model"] = raw
|
|
out["field"] = [f for g in m["groups"] for f in g["fields"]
|
|
if f["key"] == "price_service_token"]
|
|
|
|
# a model of a site with NO token: shown empty, not as 'set'
|
|
sup0 = Sup(dict(SITE, price_service_token=""))
|
|
r0, root0 = await serve(sup0)
|
|
try:
|
|
async with s.get(root0 + "config/model?lang=en") as r:
|
|
out["model_unset"] = await r.json()
|
|
finally:
|
|
await r0.cleanup()
|
|
|
|
# --- classic page save: the page posts `options`, placeholder in
|
|
opts = copy.deepcopy(m["options"])
|
|
opts["peak_cap_w"] = 4100
|
|
st, body = await post("config/save",
|
|
{"options": opts, "fingerprint": m["fingerprint"]})
|
|
out["classic_placeholder"] = (st, sup.options["price_service_token"],
|
|
sup.options["peak_cap_w"], body)
|
|
# --- classic save with the field emptied
|
|
_, raw2 = await get("config/model?lang=en")
|
|
m2 = json.loads(raw2)
|
|
opts = copy.deepcopy(m2["options"])
|
|
opts["price_service_token"] = ""
|
|
opts["peak_cap_w"] = 4200
|
|
st, body = await post("config/save",
|
|
{"options": opts, "fingerprint": m2["fingerprint"]})
|
|
out["classic_empty"] = (st, sup.options["price_service_token"],
|
|
sup.options["peak_cap_w"])
|
|
# --- classic save with the key left out
|
|
_, raw2 = await get("config/model?lang=en")
|
|
m2 = json.loads(raw2)
|
|
opts = copy.deepcopy(m2["options"])
|
|
del opts["price_service_token"]
|
|
opts["peak_cap_w"] = 4250
|
|
st, body = await post("config/save",
|
|
{"options": opts, "fingerprint": m2["fingerprint"]})
|
|
out["classic_absent"] = (st, sup.options["price_service_token"])
|
|
# --- the new settings page: model in, with and without base
|
|
_, raw3 = await get("config/model?lang=en")
|
|
m3 = json.loads(raw3)
|
|
ui = copy.deepcopy(m3["ui"])
|
|
ui["peak_cap_w"] = 4300
|
|
st, body = await post("config/save", {
|
|
"model": ui, "base": copy.deepcopy(m3["ui"]),
|
|
"fingerprint": m3["fingerprint"]})
|
|
out["ui_base"] = (st, sup.options["price_service_token"])
|
|
_, raw3 = await get("config/model?lang=en")
|
|
m3 = json.loads(raw3)
|
|
ui = copy.deepcopy(m3["ui"])
|
|
ui["peak_cap_w"] = 4400
|
|
st, body = await post("config/save",
|
|
{"model": ui, "fingerprint": m3["fingerprint"]})
|
|
out["ui_nobase"] = (st, sup.options["price_service_token"])
|
|
# --- validate with the placeholder in: no 'too short', no echo
|
|
_, raw4 = await get("config/model?lang=en")
|
|
m4 = json.loads(raw4)
|
|
st, body = await post("config/validate", {"options": m4["options"]})
|
|
out["validate_placeholder"] = (st, json.loads(body)["problems"])
|
|
st, body = await post("config/validate", {"model": m4["ui"]})
|
|
out["validate_ui_placeholder"] = (st, json.loads(body)["problems"])
|
|
# --- a refused save (token too short) and a stale one
|
|
opts = copy.deepcopy(m4["options"])
|
|
opts["price_service_token"] = "short"
|
|
st, body = await post("config/save",
|
|
{"options": opts, "fingerprint": m4["fingerprint"]})
|
|
out["refused"] = (st, json.loads(body), sup.options["price_service_token"])
|
|
opts = copy.deepcopy(m4["options"])
|
|
opts["peak_cap_w"] = 4500
|
|
st, body = await post("config/save",
|
|
{"options": opts, "fingerprint": "not-current"})
|
|
out["stale"] = (st, json.loads(body))
|
|
# --- a refusal that is not about the token
|
|
opts = copy.deepcopy(m4["options"])
|
|
opts["max_w"] = -5
|
|
st, body = await post("config/save",
|
|
{"options": opts, "fingerprint": m4["fingerprint"]})
|
|
out["refused_other"] = (st, sup.options["price_service_token"],
|
|
sup.options["max_w"])
|
|
# --- a new value replaces
|
|
_, raw5 = await get("config/model?lang=en")
|
|
m5 = json.loads(raw5)
|
|
opts = copy.deepcopy(m5["options"])
|
|
opts["price_service_token"] = NEW_TOKEN
|
|
st, body = await post("config/save",
|
|
{"options": opts, "fingerprint": m5["fingerprint"]})
|
|
out["replaced"] = (st, sup.options["price_service_token"], body)
|
|
out["writes"] = len(sup.writes)
|
|
finally:
|
|
await runner.cleanup()
|
|
out["seen"] = seen
|
|
return out
|
|
|
|
|
|
S = asyncio.run(scenario())
|
|
|
|
print("1. the token is in no page and no JSON answer")
|
|
check("GET /config (the classic page) has no token",
|
|
TOKEN not in S["page"] and NEW_TOKEN not in S["page"])
|
|
check("GET /config/model has no token anywhere in its JSON",
|
|
TOKEN not in S["raw_model"])
|
|
for key in ("options", "running", "ui"):
|
|
check(f"...and {key} shows the placeholder, not the token",
|
|
json.dumps(S["model"][key]).count(SECRET_SHOWN) >= 1
|
|
and TOKEN not in json.dumps(S["model"][key]))
|
|
check("...the placeholder stands in the token's own place",
|
|
S["model"]["options"]["price_service_token"] == SECRET_SHOWN
|
|
and S["model"]["running"]["price_service_token"] == SECRET_SHOWN
|
|
and S["model"]["ui"]["price_service_token"] == SECRET_SHOWN)
|
|
check("a site with no token shows it empty, so the page can say 'not set'",
|
|
S["model_unset"]["options"]["price_service_token"] == ""
|
|
and S["model_unset"]["ui"]["price_service_token"] == "")
|
|
check("no answer the app gave in the whole run (page, model, validate, saves, "
|
|
"refusals, the stale one) holds the stored token",
|
|
not [t for t in S["seen"] if TOKEN in t])
|
|
check("...nor the new token, after it was saved",
|
|
not [t for t in S["seen"] if NEW_TOKEN in t])
|
|
|
|
print("\n2. the classic page draws it as a password field")
|
|
check("the field model types it as a password",
|
|
[f["type"] for f in S["field"]] == ["password"])
|
|
check("the classic page's script makes a password input for that type",
|
|
"'password'" in S["page"] and "input.type = 'password'" in S["page"])
|
|
|
|
print("\n3. saves keep the stored token unless given a new one")
|
|
st, tok, cap, body = S["classic_placeholder"]
|
|
check("a classic save with the placeholder keeps the token and saves the "
|
|
"other edit", st == 200 and tok == TOKEN and cap == 4100)
|
|
st, tok, cap = S["classic_empty"]
|
|
check("a classic save with the field emptied keeps the token",
|
|
st == 200 and tok == TOKEN and cap == 4200)
|
|
check("a classic save without the key keeps the token",
|
|
S["classic_absent"] == (200, TOKEN))
|
|
check("a save from the new settings page keeps it (with a base)",
|
|
S["ui_base"] == (200, TOKEN))
|
|
check("...and without a base", S["ui_nobase"] == (200, TOKEN))
|
|
st, tok, body = S["replaced"]
|
|
check("a new value replaces the token", st == 200 and tok == NEW_TOKEN)
|
|
check("...and the answer to that save does not echo it",
|
|
NEW_TOKEN not in body)
|
|
st, probs = S["validate_placeholder"]
|
|
check("validate with the placeholder in sees the stored token: nothing about "
|
|
"the token is refused", st == 200 and not [p for p in probs if "token" in p])
|
|
st, probs = S["validate_ui_placeholder"]
|
|
check("...also from the new page's model",
|
|
st == 200 and not [p for p in probs if "token" in p])
|
|
st, body, tok = S["refused"]
|
|
check("a refused save (token too short) is 422, says why in short plain "
|
|
"words, and writes nothing",
|
|
st == 422 and tok == TOKEN
|
|
and any("too short" in p for p in body["problems"]))
|
|
st, body = S["stale"]
|
|
check("a stale refusal is 409 and its `stored` shows the placeholder",
|
|
st == 409 and body["stored"]["price_service_token"] == SECRET_SHOWN)
|
|
check("a refusal about something else (422) writes nothing, so the token and "
|
|
"the rating stay",
|
|
S["refused_other"][0] == 422 and S["refused_other"][1] == TOKEN
|
|
and S["refused_other"][2] != -5)
|
|
|
|
print("\n4. every password option, not only this name")
|
|
other = [{"name": "mqtt_secret", "type": "string", "format": "password"},
|
|
{"name": "other_pw", "type": "password"},
|
|
{"name": "plain", "type": "string"}]
|
|
keys = configui.secret_keys(other)
|
|
check("a schema password is found in both spellings, a plain string is not",
|
|
keys == {"mqtt_secret", "other_pw"})
|
|
check("the price token is counted by name too, in case a schema is missing",
|
|
configui.secret_keys([], price.SECRET_OPTIONS)
|
|
== frozenset(price.SECRET_OPTIONS) and "price_service_token" in
|
|
price.SECRET_OPTIONS)
|
|
blob = {"mqtt_secret": "hunter2hunter2", "other_pw": "", "plain": "x",
|
|
"nested": [{"mqtt_secret": "deep-secret-value"}]}
|
|
shown = configui.hide_secrets(blob, keys)
|
|
check("hide_secrets: set -> placeholder, empty -> empty, other keys alone, "
|
|
"at any depth",
|
|
shown == {"mqtt_secret": SECRET_SHOWN, "other_pw": "", "plain": "x",
|
|
"nested": [{"mqtt_secret": SECRET_SHOWN}]}
|
|
and blob["mqtt_secret"] == "hunter2hunter2")
|
|
back = configui.keep_secrets({"mqtt_secret": SECRET_SHOWN, "other_pw": "new",
|
|
"plain": "y"}, blob, keys)
|
|
check("keep_secrets: the placeholder gives the stored value, a new value "
|
|
"stays",
|
|
back["mqtt_secret"] == "hunter2hunter2" and back["other_pw"] == "new"
|
|
and back["plain"] == "y")
|
|
check("keep_secrets: a placeholder with nothing stored is never saved as "
|
|
"the placeholder",
|
|
configui.keep_secrets({"mqtt_secret": SECRET_SHOWN}, {}, keys)
|
|
["mqtt_secret"] == "")
|
|
|
|
|
|
async def other_option_scenario():
|
|
"""A second password option, found from the schema, is hidden and kept."""
|
|
schema = SCHEMA + [{"name": "extra_secret", "type": "string",
|
|
"format": "password", "required": False,
|
|
"optional": True}]
|
|
sup = Sup(dict(SITE, extra_secret="extra-secret-value-12345"), schema)
|
|
runner, root = await serve(sup)
|
|
try:
|
|
async with ClientSession() as s:
|
|
async with s.get(root + "config/model?lang=en") as r:
|
|
raw = await r.text()
|
|
m = json.loads(raw)
|
|
opts = copy.deepcopy(m["options"])
|
|
opts["peak_cap_w"] = 4100
|
|
async with s.post(root + "config/save", json={
|
|
"options": opts, "fingerprint": m["fingerprint"]}) as r:
|
|
st = r.status
|
|
return raw, st, sup.options.get("extra_secret")
|
|
finally:
|
|
await runner.cleanup()
|
|
|
|
|
|
_raw, _st, _kept = asyncio.run(other_option_scenario())
|
|
check("a password option the add-on has no name for is hidden from the model",
|
|
"extra-secret-value-12345" not in _raw)
|
|
check("...and kept by a save that sends the placeholder",
|
|
_st == 200 and _kept == "extra-secret-value-12345")
|
|
|
|
async def named_scenario():
|
|
"""A schema that does not call the token a password (a Supervisor that
|
|
spells it another way): the add-on still hides it, by name."""
|
|
schema = [dict(n, format=None) if n["name"] == "price_service_token" else n
|
|
for n in SCHEMA]
|
|
sup = Sup(SITE, schema)
|
|
runner, root = await serve(sup)
|
|
try:
|
|
async with ClientSession() as s:
|
|
async with s.get(root + "config/model?lang=en") as r:
|
|
return await r.text()
|
|
finally:
|
|
await runner.cleanup()
|
|
|
|
|
|
check("the price token is hidden by name even when the schema does not mark "
|
|
"it as a password", TOKEN not in asyncio.run(named_scenario()))
|
|
|
|
print("\n5. PRICE-01 round 3: the fixed-tariff note on the real validate route")
|
|
_PROFILE = json.dumps({"periods": [
|
|
{"name": "all", "days": "all", "from": "00:00", "to": "24:00",
|
|
"eur_per_kwh": 0.25}]})
|
|
_NOTE = {"en": ("With the price service, the fixed tariff is only used if no "
|
|
"prices are available."),
|
|
"nl": ("Met de prijsdienst wordt het vaste tarief alleen gebruikt als "
|
|
"er geen prijzen zijn.")}
|
|
|
|
|
|
async def note_scenario(site):
|
|
runner, root = await serve(Sup(site))
|
|
out = {}
|
|
try:
|
|
async with ClientSession() as s:
|
|
for lang in ("en", "nl"):
|
|
async with s.get(root + f"config/model?lang={lang}") as r:
|
|
m = await r.json()
|
|
async with s.post(root + "config/validate",
|
|
json={"model": m["ui"], "lang": lang}) as r:
|
|
b = await r.json()
|
|
out[lang] = [x for x in b.get("messages", [])
|
|
if x["text"] in _NOTE.values()]
|
|
finally:
|
|
await runner.cleanup()
|
|
return out
|
|
|
|
|
|
_with = asyncio.run(note_scenario(dict(SITE, static_profile=_PROFILE)))
|
|
check("an English page gets the note in English, as info, beside a field",
|
|
[(x["text"], x["tier"]) for x in _with["en"]] == [(_NOTE["en"], "info")]
|
|
and _with["en"][0]["field"])
|
|
check("...and a Dutch page gets it in Dutch, beside the same field",
|
|
[(x["text"], x["tier"]) for x in _with["nl"]] == [(_NOTE["nl"], "info")]
|
|
and _with["nl"][0]["field"] == _with["en"][0]["field"])
|
|
_blank = asyncio.run(note_scenario(dict(SITE, static_profile=" ")))
|
|
check("a spaces-only fixed tariff: no note on either page",
|
|
_blank == {"en": [], "nl": []})
|
|
|
|
print("\n6. the shipped config.yaml")
|
|
cfg = open(os.path.join(HERE, "config.yaml"), encoding="utf-8").read()
|
|
schema_text = cfg[cfg.index("\nschema:"):]
|
|
pw = {line.split(":")[0].strip() for line in schema_text.splitlines()
|
|
if line.split(":", 1)[-1].strip().rstrip("?") == "password"}
|
|
check(f"every password option of config.yaml is counted as a secret ({sorted(pw)})",
|
|
pw and pw <= configui.secret_keys(SCHEMA, price.SECRET_OPTIONS))
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: " + "; ".join(fails))
|
|
sys.exit(1)
|
|
print("all checks passed")
|