"""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")