fix(settings): enforce authoritative global image-gen policy and non-admin preference persistence

- Enforce authoritative administrator global image_gen_enabled disable in get_user_setting, do_generate_image, and image_editing_provider_available, preventing non-admin user preference overrides while preserving user-disabled opt-outs and can_generate_images privilege checks across direct, gallery, and editor paths.
- Filter non-admin preference persistence in saveImagePreferences to user preference keys only (image_model, image_edit_model, image_quality), preventing global policy toggles from leaking into user preferences.
- Return accurate 200 OK Response from saveImagePreferences after fallback persistence succeeds so settings UI displays 'Saved' instead of false 403 error, while retaining visibility of real errors.
- Load per-user preferences via readImagePreferences in initImageSettings so non-admin selections persist and display correctly after page reload.
- Clear per-user model preferences on successful admin save (POST /api/auth/settings) via PUT /api/prefs/{key} with value: null to prevent stale per-user preferences from shadowing updated global settings.
- Support clearing keys in PUT /api/prefs/{key} when value is null.
- Add comprehensive Python regression tests for global-disabled/user-enabled, global-enabled/user-disabled, and authorized generation, plus Node and browser regression tests for non-admin persistence, toggle isolation, reload, and admin clearing.
This commit is contained in:
Alexandre Teixeira
2026-10-31 20:35:44 +00:00
parent 1cb597ce12
commit 1888839e06
8 changed files with 467 additions and 11 deletions
+6 -2
View File
@@ -118,8 +118,12 @@ def setup_prefs_routes():
async def set_pref(request: Request, key: str, body: dict):
user = get_current_user(request)
prefs = _load_for_user(user)
prefs[key] = body.get("value")
val = body.get("value")
if val is None:
prefs.pop(key, None)
else:
prefs[key] = val
_save_for_user(user, prefs)
return {"key": key, "value": prefs[key]}
return {"key": key, "value": prefs.get(key)}
return router
+7 -3
View File
@@ -1181,9 +1181,9 @@ def image_editing_provider_available(owner: Optional[str] = None) -> bool:
request on this so it never falls back to deterministic approximations.
"""
try:
from src.settings import load_settings
from src.settings import get_user_setting, load_settings
if not load_settings().get("image_gen_enabled", True):
if not load_settings().get("image_gen_enabled", True) or not get_user_setting("image_gen_enabled", owner=owner or "", default=True):
return False
except Exception:
pass
@@ -1283,8 +1283,12 @@ async def do_generate_image(content: str, session_id: Optional[str] = None, owne
def get_user_setting(key, owner="", default=None):
return default
if not get_user_setting("image_gen_enabled", owner=owner, default=True):
# The administrator's global image-generation disable switch is authoritative.
# Non-admin preferences cannot re-enable generation when disabled globally.
if not _settings.get("image_gen_enabled", True):
return {"error": "Image generation is disabled by the administrator."}
if not get_user_setting("image_gen_enabled", owner=owner or "", default=True):
return {"error": "Image generation is disabled by user preference."}
# Use user- or admin-configured model/quality if not specified by the tool call
if not model_spec:
+14
View File
@@ -314,6 +314,20 @@ def get_user_setting(key: str, owner: str = "", default: Any = None) -> Any:
Falls back gracefully if the prefs module can't be imported (cycle/early
boot) — admin-global settings keep working.
"""
if key == "image_gen_enabled":
global_val = get_setting("image_gen_enabled", default if default is not None else False)
if not global_val:
return False
if owner:
try:
from routes.prefs_routes import _load_for_user
prefs = _load_for_user(owner) or {}
if "image_gen_enabled" in prefs and prefs["image_gen_enabled"] is not None:
return bool(prefs["image_gen_enabled"])
except Exception:
pass
return bool(global_val)
if owner and key in _PER_USER_KEYS:
try:
from routes.prefs_routes import _load_for_user
+2 -3
View File
@@ -4,7 +4,7 @@
// them from the same places, so load order and init sequence are untouched.
import { byId as el } from './dom.js';
import { saveImagePreferences as _postSettings } from './imagePreferences.js';
import { saveImagePreferences as _postSettings, readImagePreferences } from './imagePreferences.js';
import { modelCaps } from '../editor/ai-models.js';
import { sortModelIds } from '../modelSort.js';
let preferencesListener = null;
@@ -46,8 +46,7 @@ export async function initImageSettings() {
}
} catch (e) { console.warn('Failed to load models for image settings', e); }
try {
const settingsRes = await fetch('/api/auth/settings', { credentials: 'same-origin' });
const settings = await settingsRes.json();
const settings = await readImagePreferences();
if (settings.image_model) modelSel.value = settings.image_model;
if (editModelSel) editModelSel.value = settings.image_edit_model || '';
if (settings.image_quality) qualSel.value = settings.image_quality;
+20 -3
View File
@@ -1,5 +1,7 @@
import { invalidateSettings } from '../appConfig.js';
const USER_PREFERENCE_KEYS = ['image_model', 'image_edit_model', 'image_quality'];
export async function readImagePreferences() {
const [prefsRes, settingsRes] = await Promise.all([
fetch('/api/prefs', { credentials: 'same-origin' }).catch(() => null),
@@ -12,6 +14,7 @@ export async function readImagePreferences() {
...settings,
image_model: prefs?.image_model || settings?.image_model || '',
image_edit_model: prefs?.image_edit_model || settings?.image_edit_model || '',
image_quality: prefs?.image_quality || settings?.image_quality || 'medium',
};
}
@@ -24,17 +27,31 @@ export function saveImagePreferences(patch) {
headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(patch),
});
if (response.status === 403) {
for (const [key, value] of Object.entries(patch)) {
const userEntries = Object.entries(patch).filter(([k]) => USER_PREFERENCE_KEYS.includes(k));
for (const [key, value] of userEntries) {
const prefRes = await fetch(`/api/prefs/${encodeURIComponent(key)}`, {
method: 'PUT', credentials: 'same-origin',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ value }),
});
if (!prefRes.ok) {
throw new Error('Could not save image preferences');
const err = new Error(await prefRes.text().catch(() => 'Could not save image preferences'));
err.status = prefRes.status;
throw err;
}
}
} else if (!response.ok) {
response = typeof Response !== 'undefined'
? new Response(JSON.stringify({ ok: true }), { status: 200, headers: { 'Content-Type': 'application/json' } })
: { ok: true, status: 200, json: async () => ({ ok: true }), text: async () => '{"ok":true}' };
} else if (response.ok) {
await Promise.all(USER_PREFERENCE_KEYS.map(key =>
fetch(`/api/prefs/${encodeURIComponent(key)}`, {
method: 'PUT', credentials: 'same-origin',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ value: null }),
}).catch(() => null)
));
} else {
throw new Error('Could not save image settings');
}
invalidateSettings();
@@ -293,3 +293,208 @@ def test_is_background_removal_intent_grammar_and_dos_safety():
assert not _is_background_removal_intent(evil_invalid)
elapsed_invalid = time.perf_counter() - start
assert elapsed_invalid < 0.1, f"Invalid intent parsing took too long: {elapsed_invalid:.4f}s"
def test_image_generation_global_disabled_user_enabled_rejected(tmp_path, monkeypatch):
import asyncio
from fastapi import HTTPException
from routes.gallery import gallery_routes as gallery
from routes import editor_agent_routes, prefs_routes
from src import ai_interaction, settings
prefs_file = str(tmp_path / "user_prefs.json")
monkeypatch.setattr(prefs_routes, "PREFS_FILE", prefs_file)
# Admin globally disables image generation
monkeypatch.setattr(settings, "load_settings", lambda: {"image_gen_enabled": False, "image_model": "gpt-image-1"})
# User 'alice' has image_gen_enabled = True in her preferences
prefs_routes._save_for_user("alice", {"image_gen_enabled": True, "image_model": "alice-model"})
# 1. get_user_setting must be False despite user preference
assert settings.get_user_setting("image_gen_enabled", owner="alice", default=True) is False
# 2. image_editing_provider_available must be False
assert ai_interaction.image_editing_provider_available(owner="alice") is False
# 3. Direct generation must be rejected with admin disabled message
direct_res = asyncio.run(ai_interaction.do_generate_image(
json.dumps({"prompt": "a dog"}), owner="alice"
))
assert "error" in direct_res
assert "disabled by the administrator" in direct_res["error"]
# 4. Gallery route must reject with 503
app = FastAPI()
app.include_router(gallery.setup_gallery_routes())
monkeypatch.setattr(gallery, "require_privilege", lambda request, priv: "alice")
with TestClient(app) as client:
gal_res = client.post("/api/image/generate", json={"prompt": "a dog"})
assert gal_res.status_code == 503
assert "switched off" in gal_res.json()["detail"].lower()
# 5. Editor agent route must be blocked
app_editor = FastAPI()
app_editor.include_router(editor_agent_routes.setup_editor_agent_routes())
monkeypatch.setattr(editor_agent_routes, "_user", lambda request: "alice")
routes.set_editor_planner(lambda messages, owner: json.dumps({
"reply": "Gen", "steps": [{"op": "generate_image", "params": {"prompt": "a dog"}}]
}))
try:
with TestClient(app_editor) as client:
ed_res = client.post("/api/editor/agent", json={"prompt": "create a dog"})
assert ed_res.status_code == 200
data = ed_res.json()
assert data.get("blocked") is True
assert data.get("needs_provider") is True
finally:
routes.set_editor_planner(None)
def test_image_generation_global_enabled_user_disabled_rejected(tmp_path, monkeypatch):
import asyncio
from routes.gallery import gallery_routes as gallery
from routes import editor_agent_routes, prefs_routes
from src import ai_interaction, settings
prefs_file = str(tmp_path / "user_prefs.json")
monkeypatch.setattr(prefs_routes, "PREFS_FILE", prefs_file)
# Admin globally enables image generation
monkeypatch.setattr(settings, "load_settings", lambda: {"image_gen_enabled": True, "image_model": "gpt-image-1"})
# User 'bob' has image_gen_enabled = False in his preferences
prefs_routes._save_for_user("bob", {"image_gen_enabled": False, "image_model": "bob-model"})
# 1. get_user_setting must be False per user preference
assert settings.get_user_setting("image_gen_enabled", owner="bob", default=True) is False
# 2. image_editing_provider_available must be False for bob
assert ai_interaction.image_editing_provider_available(owner="bob") is False
# 3. Direct generation must be rejected with user preference message
direct_res = asyncio.run(ai_interaction.do_generate_image(
json.dumps({"prompt": "a cat"}), owner="bob"
))
assert "error" in direct_res
assert "disabled by user preference" in direct_res["error"]
# 4. Gallery route must reject with 503
app = FastAPI()
app.include_router(gallery.setup_gallery_routes())
monkeypatch.setattr(gallery, "require_privilege", lambda request, priv: "bob")
with TestClient(app) as client:
gal_res = client.post("/api/image/generate", json={"prompt": "a cat"})
assert gal_res.status_code == 503
# 5. Editor agent route must be blocked for bob
app_editor = FastAPI()
app_editor.include_router(editor_agent_routes.setup_editor_agent_routes())
monkeypatch.setattr(editor_agent_routes, "_user", lambda request: "bob")
routes.set_editor_planner(lambda messages, owner: json.dumps({
"reply": "Gen", "steps": [{"op": "generate_image", "params": {"prompt": "a cat"}}]
}))
try:
with TestClient(app_editor) as client:
ed_res = client.post("/api/editor/agent", json={"prompt": "create a cat"})
assert ed_res.status_code == 200
data = ed_res.json()
assert data.get("blocked") is True
assert data.get("needs_provider") is True
finally:
routes.set_editor_planner(None)
def test_image_generation_authorized_generation_and_privilege_checks(tmp_path, monkeypatch):
import asyncio
from fastapi import HTTPException
from routes.gallery import gallery_routes as gallery
from routes import editor_agent_routes, prefs_routes
from src import ai_interaction, settings
prefs_file = str(tmp_path / "user_prefs.json")
monkeypatch.setattr(prefs_routes, "PREFS_FILE", prefs_file)
# Admin globally enables image generation
monkeypatch.setattr(settings, "load_settings", lambda: {"image_gen_enabled": True, "image_model": "gpt-image-1"})
# User 'charlie' has image generation enabled (no disabling pref)
prefs_routes._save_for_user("charlie", {"image_model": "charlie-gen"})
assert settings.get_user_setting("image_gen_enabled", owner="charlie", default=True) is True
# Check privilege enforcement: user lacking can_generate_images is forbidden
def require_privilege_mock(request, privilege):
if privilege == "can_generate_images":
raise HTTPException(403, "Permission denied")
return "charlie"
monkeypatch.setattr("src.auth_helpers.require_privilege", require_privilege_mock)
monkeypatch.setattr(gallery, "require_privilege", require_privilege_mock)
# Gallery checks privilege and returns 403
app = FastAPI()
app.include_router(gallery.setup_gallery_routes())
with TestClient(app) as client:
gal_res = client.post("/api/image/generate", json={"prompt": "a bird"})
assert gal_res.status_code == 403
assert "Permission denied" in gal_res.text
# Editor checks privilege on AI content and raises 403
app_editor = FastAPI()
app_editor.include_router(editor_agent_routes.setup_editor_agent_routes())
monkeypatch.setattr(editor_agent_routes, "_user", lambda request: "charlie")
monkeypatch.setattr(editor_agent_routes, "_image_provider_available", lambda owner: True)
routes.set_editor_planner(lambda messages, owner: json.dumps({
"reply": "Gen", "steps": [{"op": "generate_image", "params": {"prompt": "a bird"}}]
}))
try:
with TestClient(app_editor) as client:
ed_res = client.post("/api/editor/agent", json={"prompt": "create a bird"})
assert ed_res.status_code == 403
finally:
routes.set_editor_planner(None)
# Now grant privilege: generation proceeds
monkeypatch.setattr("src.auth_helpers.require_privilege", lambda request, priv: "charlie")
monkeypatch.setattr(gallery, "require_privilege", lambda request, priv: "charlie")
seen = []
async def fake_do_generate(content, **kwargs):
seen.append(kwargs)
return {"image_url": "data:image/png;base64,dGVzdA=="}
monkeypatch.setattr(ai_interaction, "do_generate_image", fake_do_generate)
with TestClient(app) as client:
gal_ok = client.post("/api/image/generate", json={"prompt": "a bird"})
assert gal_ok.status_code == 200
assert seen[0]["owner"] == "charlie"
def test_put_pref_clears_key_when_value_is_null(tmp_path, monkeypatch):
from routes import prefs_routes
from src import settings
prefs_file = str(tmp_path / "user_prefs.json")
monkeypatch.setattr(prefs_routes, "PREFS_FILE", prefs_file)
# Set initial preference
prefs_routes._save_for_user("admin_user", {"image_model": "old-admin-model"})
assert prefs_routes._load_for_user("admin_user").get("image_model") == "old-admin-model"
app = FastAPI()
app.include_router(prefs_routes.setup_prefs_routes())
monkeypatch.setattr("src.auth_helpers.get_current_user", lambda request: "admin_user")
monkeypatch.setattr(prefs_routes, "get_current_user", lambda request: "admin_user")
with TestClient(app) as client:
res = client.put("/api/prefs/image_model", json={"value": None})
assert res.status_code == 200
assert res.json() == {"key": "image_model", "value": None}
# Verify key is deleted from user's record
user_prefs = prefs_routes._load_for_user("admin_user")
assert "image_model" not in user_prefs
# Verify get_user_setting now falls back to global setting
monkeypatch.setattr(settings, "load_settings", lambda: {"image_model": "new-global-model"})
assert settings.get_user_setting("image_model", owner="admin_user") == "new-global-model"
@@ -454,6 +454,100 @@ def test_image_model_settings_sync_both_directions_and_survive_reopening(page):
assert len(writes) == 2
def test_non_admin_image_model_preferences_persist_and_reload(page):
global_settings = {
'image_model': 'global-default-gen',
'image_edit_model': '',
'image_quality': 'medium',
'image_gen_enabled': True,
}
user_prefs = {}
auth_posts = []
pref_puts = []
is_admin = False
def handle_auth_settings(route):
if route.request.method == 'POST':
auth_posts.append(route.request.post_data_json)
if not is_admin:
route.fulfill(status=403, json={'error': 'Admin access is required to change these settings.'})
return
global_settings.update(route.request.post_data_json)
route.fulfill(status=200, json=dict(global_settings))
return
route.fulfill(status=200, json=dict(global_settings))
def handle_prefs(route):
url = route.request.url
if route.request.method == 'PUT':
key = url.split('/api/prefs/')[-1]
data = route.request.post_data_json or {}
val = data.get('value')
pref_puts.append((key, val))
if val is None:
user_prefs.pop(key, None)
else:
user_prefs[key] = val
route.fulfill(status=200, json={'key': key, 'value': val})
return
route.fulfill(status=200, json=dict(user_prefs))
page.route('**/api/auth/settings', handle_auth_settings)
page.route('**/api/prefs**', handle_prefs)
page.route('**/api/model-endpoints', lambda route: route.fulfill(json=[{
'is_enabled': True, 'online': True, 'model_type': 'llm',
'models': ['global-default-gen', 'openai/gpt-5-image', 'openai/gpt-image-2'],
}]))
# Step 1: Non-admin loads image settings, selects a preference
page.evaluate("""async () => {
document.body.innerHTML = '<select id="set-imgModelSelect"><option value="">Auto</option></select>'
+ '<select id="set-imgEditModelSelect"><option value="">Same</option></select>'
+ '<select id="set-imgQualitySelect"><option value="medium">Medium</option></select>'
+ '<input type="checkbox" id="set-imgEnabledToggle" checked><span id="set-imgSettingsMsg"></span>';
const {initImageSettings} = await import('/static/js/settings/imageModels.js');
await initImageSettings();
}""")
# Non-admin changes generation model
page.select_option('#set-imgModelSelect', 'openai/gpt-image-2')
# Wait for status message to show 'Saved'
page.wait_for_function("document.getElementById('set-imgSettingsMsg').textContent === 'Saved'")
msg_color = page.evaluate("document.getElementById('set-imgSettingsMsg').style.color")
assert msg_color == 'var(--fg)', f"Expected var(--fg), got {msg_color}"
# Verify per-user preference was saved and global toggle was NOT propagated
assert user_prefs.get('image_model') == 'openai/gpt-image-2'
assert 'image_gen_enabled' not in user_prefs, "Global policy toggle must not be saved to user prefs"
# Step 2: Simulate page reload / re-initialization for non-admin
page.evaluate("""async () => {
document.getElementById('set-imgModelSelect').value = '';
const {initImageSettings} = await import('/static/js/settings/imageModels.js');
await initImageSettings();
}""")
page.wait_for_function("document.getElementById('set-imgModelSelect').value === 'openai/gpt-image-2'")
assert page.evaluate("document.getElementById('set-imgModelSelect').value") == 'openai/gpt-image-2'
# Step 3: Admin saves settings - pre-existing per-user preferences must be cleared
is_admin = True
page.evaluate("""async () => {
const {saveImagePreferences} = await import('/static/js/settings/imagePreferences.js');
await saveImagePreferences({ image_model: 'admin-chosen-model', image_edit_model: '', image_quality: 'high', image_gen_enabled: true });
}""")
# Verify per-user preference for image_model was cleared (PUT with value: null)
assert 'image_model' not in user_prefs, "Admin save must clear pre-existing per-user preference"
assert global_settings['image_model'] == 'admin-chosen-model'
# Admin reloads settings - receives newly saved global setting, not stale pref
page.evaluate("""async () => {
const {initImageSettings} = await import('/static/js/settings/imageModels.js');
await initImageSettings();
}""")
assert global_settings['image_model'] == 'admin-chosen-model'
def test_generation_places_result_without_stretching_or_refresh(page):
result = page.evaluate("""async () => {
const {createImageGenerationAction} = await import('/static/js/editor/image-generation.js');
+119
View File
@@ -118,3 +118,122 @@ def test_failed_save_shows_failure_in_simulated_dom():
assert data["serverErrResult"] == {"text": "Failed to save", "color": "var(--red)"}
assert data["okResult"] == {"text": "Saved", "color": "var(--fg)"}
@pytest.mark.skipif(not _HAS_NODE, reason="node binary not on PATH")
def test_image_preferences_js_policy_and_non_admin_fallback():
"""Verify imagePreferences.js non-admin fallback, policy toggle isolation, and admin clearing."""
js = """
globalThis.window = { dispatchEvent: () => {} };
globalThis.CustomEvent = class CustomEvent {};
const { saveImagePreferences, readImagePreferences } = await import('./static/js/settings/imagePreferences.js');
const results = {};
// Scenario 1: Non-admin save (POST /api/auth/settings returns 403)
const prefCalls = [];
globalThis.fetch = async (url, opts = {}) => {
if (url === '/api/auth/settings') {
return { status: 403, ok: false, text: async () => 'Forbidden' };
}
if (url.startsWith('/api/prefs/')) {
prefCalls.push({ url, body: JSON.parse(opts.body) });
return { status: 200, ok: true, json: async () => ({ ok: true }) };
}
throw new Error('Unexpected URL: ' + url);
};
const patch = {
image_gen_enabled: true,
image_model: 'user-gen-model',
image_edit_model: 'user-edit-model',
image_quality: 'high',
};
const res = await saveImagePreferences(patch);
results.fallbackResOk = res.ok;
results.fallbackResStatus = res.status;
results.prefCalls = prefCalls;
// Scenario 2: Non-admin save fails on prefs PUT (keeps actual failure visible)
globalThis.fetch = async (url) => {
if (url === '/api/auth/settings') return { status: 403, ok: false };
if (url.startsWith('/api/prefs/')) return { status: 500, ok: false, text: async () => 'Disk error' };
throw new Error('Unexpected URL: ' + url);
};
try {
await saveImagePreferences({ image_model: 'fail-model' });
results.prefFailureThrew = false;
} catch (err) {
results.prefFailureThrew = true;
results.prefFailureStatus = err.status;
}
// Scenario 3: Admin save clears pre-existing per-user preferences
const adminPrefClears = [];
globalThis.fetch = async (url, opts = {}) => {
if (url === '/api/auth/settings') {
return { status: 200, ok: true, json: async () => ({ ok: true }) };
}
if (url.startsWith('/api/prefs/')) {
adminPrefClears.push({ url, body: JSON.parse(opts.body) });
return { status: 200, ok: true, json: async () => ({ ok: true }) };
}
throw new Error('Unexpected URL: ' + url);
};
const adminRes = await saveImagePreferences({ image_model: 'admin-model', image_gen_enabled: true });
results.adminResOk = adminRes.ok;
results.adminPrefClears = adminPrefClears;
// Scenario 4: readImagePreferences loads image_quality along with models
globalThis.fetch = async (url) => {
if (url === '/api/prefs') {
return {
ok: true,
json: async () => ({ image_model: 'pref-gen', image_quality: 'low' }),
};
}
if (url === '/api/auth/settings') {
return {
ok: true,
json: async () => ({ image_model: 'global-gen', image_edit_model: 'global-edit', image_quality: 'medium' }),
};
}
throw new Error('Unexpected URL: ' + url);
};
const loaded = await readImagePreferences();
results.loaded = loaded;
console.log(JSON.stringify(results));
"""
proc = subprocess.run(["node", "--input-type=module", "-e", js], capture_output=True, text=True, check=True)
import json
data = json.loads(proc.stdout)
# 1. Non-admin fallback returns ok=True, status=200
assert data["fallbackResOk"] is True
assert data["fallbackResStatus"] == 200
# 2. image_gen_enabled was filtered out; only image_model, image_edit_model, image_quality written
written_keys = [c["url"].split("/api/prefs/")[1] for c in data["prefCalls"]]
assert "image_gen_enabled" not in written_keys
assert "image_model" in written_keys
assert "image_edit_model" in written_keys
assert "image_quality" in written_keys
# 3. Preference failure threw error and preserved status
assert data["prefFailureThrew"] is True
assert data["prefFailureStatus"] == 500
# 4. Admin save cleared per-user preferences with value: null
assert data["adminResOk"] is True
cleared_keys = [c["url"].split("/api/prefs/")[1] for c in data["adminPrefClears"]]
assert "image_model" in cleared_keys
for call in data["adminPrefClears"]:
assert call["body"] == {"value": None}
# 5. readImagePreferences includes image_quality from prefs
assert data["loaded"]["image_model"] == "pref-gen"
assert data["loaded"]["image_edit_model"] == "global-edit"
assert data["loaded"]["image_quality"] == "low"