Fix audit findings in queue fairness and official account lifecycle

docs/audits/2026-10-05-project-audit.md

@@ -0,0 +1,83 @@
+# Frame Flow 项目审计 — 2026-10-05
+
+审计基线：794a7e1485cc9046be3e77f6557c75740305ac47。
+范围：本地源码、配置、Mock 测试与依赖审计；不包含本次 NAS 状态验证、真实账号登录或真实设备写入。未修改业务代码。
+
+## 发现
+
+### P1：队列扫描上限导致后续高优先级告警饥饿
+
+`display/service.py:222` 只取按创建时间排序的前 50 个 QUEUED 操作，再跳过
+被 override 阻挡的低优先级请求。如果这 50 项全被阻挡，第 51 项更高优先级
+告警每一轮都无法被扫描到；其他设备请求与本地渲染也可能受影响。
+
+Mock 复现：激活 priority=50 的一小时 override，排入 50 个 priority=10
+请求，再排 priority=90 请求，调用 process_one 后后者仍为 QUEUED。
+建议先筛选可执行请求，或分页扫描直到找到可执行操作；增加跨设备与优先级回归测试。
+
+### P2：会话失效未更新连接状态，并误清第一方登录
+
+`integrations/kokonna_official/rpc.py:44-51` 在官方返回“请重新登陆”时抛出
+UNAUTHORIZED，但没有使官方 session 失效。`connection.py:104-113` 只依据
+本地 token 非空，继续显示 authorized=true。`web/app.js:17` 将任意 API 401
+都当作 Frame Flow 登录失效并执行 forgetLogin，包括外部官方会话过期。
+
+Mock 复现：RPC 返回 HTTP 500 / error -1 / 请重新登陆后，authorized 仍 true。
+建议区分第一方认证与外部账号过期，持久化失效状态，保留 Frame Flow 登录，
+在官方区域提供重新连接提示。
+
+### P2：损坏的可选官方会话文件阻止整个服务启动
+
+`connection.py:44-51` 在构造函数中直接解析 JSON 并索引字段，没有异常隔离。
+DeviceRegistry 在应用启动时构建该对象。损坏 JSON、缺字段或文件不可读会
+使可选功能故障升级为全服务启动失败，影响 API Key、UI 和 Scheduler。
+
+Mock 复现：官方 secret 文件内容为 `{broken`，create_app 抛 JSONDecodeError。
+建议隔离并禁用该连接，记录不含文件内容的诊断；其他核心功能继续启动。
+
+### P2：持久化失败导致断开状态与重启状态不同
+
+`connection.py:122-127,146-157` 先修改内存，再保存。磁盘满、权限错误等
+导致保存失败时没有回滚。尤其 disconnect 已清内存，而磁盘仍保留旧 token，
+重启后又恢复旧连接；虽然原请求返回错误，随后状态查询却会显示未连接。
+
+Mock 复现：让 _save 抛 OSError；当前 authorized=false，重新构造连接后 true。
+建议先原子持久化候选状态，再提交内存状态；失败时保持一致并提供明确错误。
+
+### P2：Request ID 脱敏遗漏官方会话
+
+`api/app.py:148-150` 仅检查 API Key、Frame Flow Token、registry.keys，
+没有检查用于官方会话的 settings.additional_secrets。符合 Request ID 长度和
+字符限制的会话值若被客户端误用为 X-Request-ID，可被响应头和错误模型原样回显。
+本问题不表示未授权者能凭空取到 token，而是违反已有 secret 不回显约束。
+
+Mock 复现：完成模拟官方连接后，将 synthetic session `mock-session` 作为
+X-Request-ID 发给 /health，响应原样返回该值。建议统一使用完整 secret 集合。
+
+## 验证
+
+- pytest：154 passed，0 failed，0 warning。
+- Ruff：通过（应用和 tests）。
+- mypy：36 个源码文件通过。
+- pip-audit（requirements.lock）：No known vulnerabilities found。
+- 五项专项 Mock 复现均确认上述行为，不调用真实 KoKonna。
+- 没有因发现问题而启用官方写方法；严格 RPC allowlist 仍阻止未启用方法。
+
+## 边界与下一步
+
+测试全绿不代表无缺陷：现有套件没有覆盖以上失败场景。
+本轮没有修复或部署；建议先修复队列饥饿，再统一外部账号状态与第一方认证，
+然后补齐会话文件容错、持久化事务一致性和 Request ID 脱敏测试。
+依赖审计仅表示本次查询未发现已知漏洞，不是全部依赖安全的证明。
+
+## 修复与复验
+
+上述五项已修复：队列使用 keyset 分页查找可执行操作；官方账号失效使用
+独立 `OFFICIAL_AUTH_REQUIRED` 错误码并清除会话，前端保留第一方登录且提供
+重新连接入口；可选 secret 解析失败不再阻止核心启动；持久化候选状态成功后
+才提交内存变更；Request ID 检查涵盖官方会话凭据。
+
+复验：164 passed、0 warning（新增 10 项回归测试）；Ruff 通过；mypy 36 文件
+通过；113 项布局检查通过；官方过期/第一方登录隔离浏览器检查通过。
+已知本地凭据对照历史扫描通过（278 个 blob）；该扫描不等同于发现所有未知
+secret 的通用审计。本轮不部署 NAS，不执行真实设备写接口。

pigway_frame_flow/api/app.py

@@ -147,7 +147,12 @@ def create_app(settings=None, start_worker=True):
         )
         if any(
             s and s in request.state.request_id
-            for s in [settings.api_key, settings.api_token, *registry.keys]
+            for s in [
+                settings.api_key,
+                settings.api_token,
+                *registry.keys,
+                *settings.additional_secrets,
+            ]
         ):
             request.state.request_id = uuid4().hex
         if request.method in ["POST", "PUT", "DELETE"]:

pigway_frame_flow/display/service.py

@@ -217,11 +217,23 @@ class DisplayService:
             (device_id, now_iso()),
         )
 
+    def queued_operations(self):
+        cursor_time, cursor_id = "", ""
+        while True:
+            rows = self.db.all(
+                "SELECT * FROM operations WHERE status='QUEUED' AND "
+                "(created_at>? OR (created_at=? AND id>?)) "
+                "ORDER BY created_at,id LIMIT 50",
+                (cursor_time, cursor_time, cursor_id),
+            )
+            if not rows:
+                return
+            yield from rows
+            cursor_time, cursor_id = rows[-1]["created_at"], rows[-1]["id"]
+
     def process_one(self):
         with self.lock:
-            for op in self.db.all(
-                "SELECT * FROM operations WHERE status='QUEUED' ORDER BY created_at LIMIT 50"
-            ):
+            for op in self.queued_operations():
                 if op["action"] not in ("render", "local-upload") and self.blocked(
                     op["device_id"]
                 ):

pigway_frame_flow/errors.py

@@ -3,6 +3,7 @@ from enum import StrEnum
 
 class ErrorCode(StrEnum):
     INVALID_REQUEST = "INVALID_REQUEST"
+    OFFICIAL_AUTH_REQUIRED = "OFFICIAL_AUTH_REQUIRED"
     UNAUTHORIZED = "UNAUTHORIZED"
     FORBIDDEN = "FORBIDDEN"
     DEVICE_NOT_FOUND = "DEVICE_NOT_FOUND"

pigway_frame_flow/integrations/kokonna_official/connection.py

@@ -1,6 +1,7 @@
 """Admin account connection; password transient, session persisted as a secret."""
 
 import json
+import logging
 import os
 import tempfile
 import threading
@@ -44,14 +45,32 @@ class Connection:
         self.reader, self.path, self.secrets = reader, path, secrets
         self.username = ""
         self.lock = threading.RLock()
-        if path.exists():
-            data = json.loads(path.read_text())
-            reader.rpc = OfficialRPC(data["token"])
-            reader._robot_id = data["robot_id"]
-            self.username = data["username"]
-            self.secrets.append(data["token"])
+        try:
+            if path.exists():
+                data = json.loads(path.read_text())
+                if (
+                    not isinstance(data, dict)
+                    or not isinstance(data.get("token"), str)
+                    or not isinstance(data.get("username"), str)
+                    or (
+                        data.get("robot_id") is not None
+                        and (type(data["robot_id"]) is not int or data["robot_id"] <= 0)
+                    )
+                    or "robot_id" not in data
+                ):
+                    raise ValueError()
+                reader.rpc = OfficialRPC(data["token"])
+                reader._robot_id = data["robot_id"]
+                self.username = data["username"]
+                self.secrets.append(data["token"])
+        except (OSError, ValueError, KeyError, TypeError):
+            reader.rpc.disconnect()
+            reader._robot_id = None
+            logging.getLogger("frame-flow").warning(
+                "Official connection configuration unavailable; reconnect required"
+            )
 
-    def _save(self):
+    def _save(self, token, username, robot_id):
         self.path.parent.mkdir(parents=True, exist_ok=True)
         with tempfile.NamedTemporaryFile(
             mode="w", dir=self.path.parent, delete=False
@@ -60,9 +79,9 @@ class Connection:
             try:
                 json.dump(
                     {
-                        "token": self.reader.rpc._token,
-                        "username": self.username,
-                        "robot_id": self.reader._robot_id,
+                        "token": token,
+                        "username": username,
+                        "robot_id": robot_id,
                     },
                     f,
                 )
@@ -73,6 +92,31 @@ class Connection:
             finally:
                 temp.unlink(missing_ok=True)
 
+    def _commit(self, rpc, username, robot_id):
+        self._save(rpc._token, username, robot_id)
+        self.reader.rpc = rpc
+        self.reader._robot_id = robot_id
+        self.reader._cached = None
+        self.username = username
+        if rpc._token and rpc._token not in self.secrets:
+            self.secrets.append(rpc._token)
+
+    def _query(self, call):
+        try:
+            return call()
+        except FlowError as error:
+            if error.code == E.OFFICIAL_AUTH_REQUIRED:
+                self.reader.rpc.disconnect()
+                self.reader._cached = None
+                self.reader._robot_id = None
+                try:
+                    self._save("", self.username, None)
+                except OSError:
+                    logging.getLogger("frame-flow").warning(
+                        "Expired official connection could not be persisted"
+                    )
+            raise
+
     def _devices(self, rpc):
         if self.reader.mock:
             return [
@@ -85,7 +129,9 @@ class Connection:
                 )
             ]
         try:
-            data = rpc.devices()["data"]
+            data = (
+                self._query(rpc.devices) if rpc is self.reader.rpc else rpc.devices()
+            )["data"]
             return [
                 OfficialDevice(
                     id=row["id"],
@@ -119,12 +165,7 @@ class Connection:
             if not self.reader.mock:
                 rpc.login(username, password)
             devices = self._devices(rpc)
-            self.reader.rpc = rpc
-            self.reader._robot_id = None
-            self.reader._cached = None
-            self.username = username
-            self.secrets.append(rpc._token)
-            self._save()
+            self._commit(rpc, username, None)
             return ConnectionState(
                 authorized=True,
                 username=username,
@@ -143,20 +184,14 @@ class Connection:
                 raise FlowError(
                     E.DEVICE_NOT_FOUND, "Official device not in this account", 404
                 )
-            self.reader._robot_id = robot_id
-            self.reader._cached = None
-            self._save()
+            self._commit(self.reader.rpc, self.username, robot_id)
             return self.state()
 
     def disconnect(self):
         with self.lock:
-            self.reader.rpc.disconnect()
-            self.reader._robot_id = None
-            self.reader._cached = None
-            self.username = ""
-            self._save()
+            self._commit(OfficialRPC(), "", None)
             return self.state()
 
     def widgets(self):
         with self.lock:
-            return self.reader.read()
+            return self._query(self.reader.read)

pigway_frame_flow/integrations/kokonna_official/rpc.py

@@ -42,13 +42,19 @@ class OfficialRPC:
                 allow_redirects=False,
             )
             if response.status_code in (401, 403):
+                self.disconnect()
                 raise FlowError(
-                    E.UNAUTHORIZED, "Official account authentication required", 401
+                    E.OFFICIAL_AUTH_REQUIRED,
+                    "Official account authentication required",
+                    401,
                 )
             payload = response.json()
             error = payload.get("error")
             if isinstance(error, dict) and error.get("message") == "请重新登陆":
-                raise FlowError(E.UNAUTHORIZED, "Official account session expired", 401)
+                self.disconnect()
+                raise FlowError(
+                    E.OFFICIAL_AUTH_REQUIRED, "Official account session expired", 401
+                )
             response.raise_for_status()
             if error or not isinstance(payload.get("result"), dict):
                 raise ValueError()

pigway_frame_flow/web/app.js

@@ -14,7 +14,7 @@ async function api(path,options={}){
  const headers={...(state.csrf?{'X-CSRF-Token':state.csrf}:{}),...options.headers};
  if(options.body&&!(options.body instanceof FormData)){headers['Content-Type']='application/json';options.body=JSON.stringify(options.body);}
  const response=await fetch(path,{...options,headers});
- if(!response.ok){if(response.status===401)forgetLogin();let error;try{error=(await response.json()).error;}catch{}
+ if(!response.ok){let error;try{error=(await response.json()).error;}catch{}if(response.status===401&&error?.code!=='OFFICIAL_AUTH_REQUIRED')forgetLogin();
  const e=new Error((error?.code||response.status)+': '+(error?.code&&t(error.code)!==error.code?t(error.code):(error?.message||t('error'))));e.code=error?.code;throw e;}
  if(response.status===204)return null;
  return options.binary?await response.blob():await response.json();

pigway_frame_flow/web/features.js

@@ -85,7 +85,7 @@ async function officialContentInfo(){
   info.innerHTML=`<summary>${zh?'官方小组件':'Official widgets'} · ${data.catalog.length}${!data.configured?(zh?' · 需官方账号授权':' · Official account required'):''}</summary><p>${esc(label)}</p><p class="notice">${esc(currentLabel)}</p><p class="notice widget-auth-notice">${zh?'如需使用官方小组件，需要另外授权 KoKonna 官方账号登录信息。现有 API Key 接入继续用于设备状态、图片和图库，不受影响。':'Official widgets require separate KoKonna account authorization. Your existing API Key continues to support device status, images and gallery.'}</p><p>${zh?'此入口只读，不会切换或取消官方订阅。':'Read only. This does not change official subscriptions.'}</p><input type="search" aria-label="${zh?'搜索小组件':'Search widgets'}" placeholder="${zh?'搜索名称或分类':'Search name or category'}"><div class="widget-list"></div>`;
   const render=()=>{const query=info.querySelector('input').value.toLowerCase();info.querySelector('.widget-list').innerHTML=data.catalog.filter(x=>(x.name+' '+x.tags.join(' ')).toLowerCase().includes(query)).map(x=>`<article class="card"><strong>${esc(x.name)}</strong><p>${esc(x.tags.join(' · '))}</p><small>${esc(x.arg_keys.join(' · '))}</small></article>`).join('');};
   info.querySelector('input').oninput=render;render();officialConnectionPanel(info);
- }catch(e){info.querySelector('p').textContent=e.message;}
+ }catch(e){info.innerHTML=`<summary>${zh?'官方小组件':'Official widgets'}</summary><p class="notice widget-auth-notice">${esc(e.message)}</p>`;officialConnectionPanel(info);}
 }
 
 async function officialConnectionPanel(info){

scripts/ui-audit-regressions.cjs

@@ -0,0 +1,3 @@
+/* Mock vendor expiration must preserve first-party login and show reconnect. */
+const assert=require('assert'),{chromium}=require('playwright');
+(async()=>{const browser=await chromium.launch({headless:true,executablePath:'/Applications/Google Chrome.app/Contents/MacOS/Google Chrome'});const page=await browser.newPage();const errors=[];page.on('pageerror',e=>errors.push(e.message));await page.goto('http://127.0.0.1:18787');await page.locator('#username').fill(process.env.FRAME_FLOW_TEST_USERNAME);await page.locator('#password').fill(process.env.FRAME_FLOW_TEST_PASSWORD);await page.locator('#login-form button').click();await page.locator('.hero').waitFor();assert(await page.evaluate(()=>state.settings.mock));await page.route('**/api/v1/devices/*/widgets',r=>r.fulfill({status:401,json:{error:{code:'OFFICIAL_AUTH_REQUIRED',message:'Official account session expired',request_id:'fixture',details:{}}}}));await page.route('**/api/v1/devices/*/official/connection',r=>r.fulfill({json:{authorized:false,username:'fixture',robot_id:null,mock:true,devices:[]}}));await page.locator('[data-page="templates"]').click();await page.locator('.official-content-info summary').click();await page.locator('#official-login-form').waitFor();assert(await page.evaluate(()=>Boolean(state.token&&state.csrf)));assert(await page.evaluate(async()=>Boolean((await api('/api/v1/auth/me')).username)));await page.route('**/api/v1/test-first-party-expired',r=>r.fulfill({status:401,json:{error:{code:'UNAUTHORIZED',message:'Expired',request_id:'fixture',details:{}}}}));await page.evaluate(async()=>{try{await api('/api/v1/test-first-party-expired');}catch{}});assert(await page.evaluate(()=>state.token===''));assert(errors.length===0);await browser.close();console.log('Audit UI: official expiry preserves login, reconnect visible, own expiry clears login; PASS');})().catch(e=>{console.error(e.message);process.exit(1)});

tests/test_audit_regressions.py

@@ -0,0 +1,103 @@
+import json
+from unittest.mock import Mock, patch
+
+import pytest
+
+from pigway_frame_flow.api.app import create_app
+from pigway_frame_flow.errors import ErrorCode, FlowError
+from pigway_frame_flow.integrations.kokonna_official.connection import Connection
+from pigway_frame_flow.integrations.kokonna_official.widgets import WidgetReader
+from pigway_frame_flow.models import DisplayRequest
+
+
+def enqueue(service, priority):
+    payload = DisplayRequest(
+        content={"type": "template", "template": "quote", "data": {}},
+        options={"priority": priority, "duration": 3600, "restore_previous": True},
+    )
+    return service.submit(
+        "display", payload.model_dump(mode="json"), "living-room", "admin", "audit"
+    )
+
+
+def test_later_alert_is_not_starved_by_first_fifty_blocked_jobs(service):
+    enqueue(service, 50)
+    service.process_one()
+    for _ in range(51):
+        enqueue(service, 10)
+    high = enqueue(service, 90)
+    assert service.process_one()
+    assert service.operation(high)["status"] == "SYNCING"
+    assert service.registry.get("living-room").client.uploads == 2
+
+
+@pytest.mark.parametrize(
+    "content",
+    [
+        "{broken",
+        "{}",
+        '{"token":123,"username":"user","robot_id":1}',
+        '{"token":"secret","username":"user","robot_id":true}',
+    ],
+)
+def test_broken_optional_secret_does_not_block_core_start(settings, content, caplog):
+    settings.config_dir.mkdir(parents=True)
+    (settings.config_dir / "official-living-room.secret").write_text(content)
+    app = create_app(settings, start_worker=False)
+    registry = app.state.service.registry
+    assert not registry.official["living-room"].state().authorized
+    assert registry.get("living-room").device().status == "online"
+    assert "secret" not in caplog.text
+
+
+@pytest.mark.parametrize("action", ["disconnect", "select", "login"])
+def test_disk_failure_preserves_existing_connection(settings, action):
+    path = settings.config_dir / "official-test.secret"
+    conn = Connection(WidgetReader(mock=True), path, [])
+    conn.login("old-user", "not-saved")
+    if action != "select":
+        conn.select(1)
+    before = path.read_text()
+    with patch.object(conn, "_save", side_effect=OSError("fixture disk failure")):
+        with pytest.raises(OSError):
+            if action == "login":
+                conn.login("new-user", "not-saved")
+            elif action == "select":
+                conn.select(1)
+            else:
+                conn.disconnect()
+    assert conn.state().authorized
+    assert conn.state().username == "old-user"
+    assert conn.state().robot_id == (None if action == "select" else 1)
+    assert path.read_text() == before
+    assert Connection(WidgetReader(mock=True), path, []).state().authorized
+
+
+def test_expired_session_invalidated_in_memory_and_on_disk(settings):
+    path = settings.config_dir / "official-test.secret"
+    conn = Connection(WidgetReader(), path, [])
+    conn._commit(conn.reader.rpc.__class__("expired-session"), "user", 1)
+    response = Mock(status_code=500)
+    response.json.return_value = {"error": {"code": -1, "message": "请重新登陆"}}
+    with (
+        patch("requests.post", return_value=response),
+        pytest.raises(FlowError) as error,
+    ):
+        conn.widgets()
+    assert error.value.code == ErrorCode.OFFICIAL_AUTH_REQUIRED
+    assert not conn.state().authorized
+    assert json.loads(path.read_text())["token"] == ""
+    assert not Connection(WidgetReader(), path, []).state().authorized
+
+
+def test_official_session_not_reflected_in_request_id(client):
+    client.post(
+        "/api/v1/devices/living-room/official/connection",
+        json={"username": "fixture", "password": "fixture"},
+    )
+    response = client.get("/health", headers={"X-Request-ID": "mock-session"})
+    assert response.headers["X-Request-ID"] != "mock-session"
+    error = client.get(
+        "/api/v1/devices/missing", headers={"X-Request-ID": "mock-session"}
+    )
+    assert error.json()["error"]["request_id"] != "mock-session"

tests/test_official_module.py

@@ -52,7 +52,7 @@ def test_rpc_http_500_relogin_is_auth_error_not_cloud_failure():
         pytest.raises(FlowError) as error,
     ):
         OfficialRPC("secret").devices()
-    assert error.value.code == ErrorCode.UNAUTHORIZED
+    assert error.value.code == ErrorCode.OFFICIAL_AUTH_REQUIRED
     assert "secret" not in str(error.value)