fix(settings): a plugin's password is sealed at rest, and a blank box keeps it (#5671)
The UniFi password sat in plugin.unifi in plain text: SECRET_KEYS covers whole settings, and a plugin's settings are one dict. Saving the plugin's settings page also wiped it, since the password box is never filled back in and its blank value was stored. - A plugin field whose name holds password, secret or token is a credential (is_plugin_secret). set_setting seals it inside the dict, every read opens it, startup seals any plaintext left from before, and the undecryptable-secrets banner names it as plugin.<name>.<field>. - The settings page uses the same test for its password boxes, and a blank box keeps the saved value. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,91 @@
|
||||
"""Integration: a plugin's password is ciphertext in app_settings, reads back
|
||||
as itself, is sealed at startup if it was stored in plain text, and is kept
|
||||
when the settings form is saved with its box blank (#5671). Requires
|
||||
STEWARD_DATABASE_URL."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app():
|
||||
if not os.environ.get("STEWARD_DATABASE_URL"):
|
||||
pytest.skip("needs Postgres")
|
||||
from steward.app import create_app
|
||||
return create_app(testing=False)
|
||||
|
||||
|
||||
def test_a_plugin_password_is_sealed_read_back_and_kept_on_a_blank_save(app):
|
||||
from sqlalchemy import delete, select
|
||||
|
||||
from steward.core.crypto import is_encrypted
|
||||
from steward.core.settings import get_setting, migrate_plaintext_secrets, set_setting
|
||||
from steward.models.settings import AppSetting
|
||||
from steward.models.users import User, UserRole
|
||||
|
||||
async def raw(key):
|
||||
async with app.db_sessionmaker() as s:
|
||||
row = (await s.execute(select(AppSetting).where(AppSetting.key == key))).scalar_one()
|
||||
return json.loads(row.value_json)
|
||||
|
||||
async def go():
|
||||
tag = uuid.uuid4().hex[:8]
|
||||
cfg = {"enabled": False, "host": "https://10.9.9.1", "username": "admin",
|
||||
"password": f"pw-{tag}", "site": "default", "verify_ssl": False,
|
||||
"poll_interval_seconds": 60, "top_clients": 10}
|
||||
async with app.db_sessionmaker() as s:
|
||||
async with s.begin():
|
||||
await set_setting(s, "plugin.unifi", cfg)
|
||||
stored = await raw("plugin.unifi")
|
||||
assert is_encrypted(stored["password"]) and stored["username"] == "admin"
|
||||
async with app.db_sessionmaker() as s:
|
||||
assert (await get_setting(s, "plugin.unifi"))["password"] == f"pw-{tag}"
|
||||
|
||||
# A row written before this fix, in plain text, is sealed at startup.
|
||||
async with app.db_sessionmaker() as s:
|
||||
async with s.begin():
|
||||
row = (await s.execute(select(AppSetting).where(AppSetting.key == "plugin.unifi"))).scalar_one()
|
||||
row.value_json = json.dumps(cfg)
|
||||
assert not is_encrypted((await raw("plugin.unifi"))["password"])
|
||||
await asyncio.to_thread(migrate_plaintext_secrets, app.config["DATABASE_URL"])
|
||||
assert is_encrypted((await raw("plugin.unifi"))["password"])
|
||||
|
||||
# Saving the settings page with the password box blank keeps it.
|
||||
async with app.db_sessionmaker() as s:
|
||||
async with s.begin():
|
||||
user = User(username=f"ps-{tag}", email=f"ps-{tag}@test", password_hash="x",
|
||||
role=UserRole.admin, is_active=True)
|
||||
s.add(user)
|
||||
uid = user.id
|
||||
client = app.test_client()
|
||||
async with client.session_transaction() as sess:
|
||||
sess.update(user_id=uid, user_role="admin", username=f"ps-{tag}",
|
||||
_csrf="test-token", login_at=time.time())
|
||||
resp = await client.post("/settings/plugins/unifi/", form={
|
||||
"csrf_token": "test-token", "plugin.unifi.host": "https://10.9.9.1",
|
||||
"plugin.unifi.username": "admin", "plugin.unifi.password": "",
|
||||
"plugin.unifi.site": "default", "plugin.unifi.poll_interval_seconds": "120",
|
||||
"plugin.unifi.top_clients": "10"})
|
||||
assert resp.status_code in (200, 302, 303), await resp.get_data(as_text=True)
|
||||
async with app.db_sessionmaker() as s:
|
||||
after = await get_setting(s, "plugin.unifi")
|
||||
assert after["password"] == f"pw-{tag}" and after["poll_interval_seconds"] == 120
|
||||
assert is_encrypted((await raw("plugin.unifi"))["password"])
|
||||
|
||||
page = await (await client.get("/settings/plugins/unifi/")).get_data(as_text=True)
|
||||
assert f"pw-{tag}" not in page and "kept; type to change" in page
|
||||
|
||||
async with app.db_sessionmaker() as s:
|
||||
async with s.begin():
|
||||
await s.execute(delete(AppSetting).where(AppSetting.key == "plugin.unifi"))
|
||||
await s.execute(delete(User).where(User.id == uid))
|
||||
|
||||
asyncio.run(go())
|
||||
Reference in New Issue
Block a user